b3nw/eve-wiki-system
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Initialize system repository: agents, infra, and configuration

Browse Source
This commit is contained in:
b3nw
2026-04-19 04:51:17 +00:00
parent b861f385e9
commit 0cb48b94e2
23 changed files with 1960 additions and 39 deletions
Download Patch File Download Diff File Expand all files Collapse all files

7
.gitignore vendored
View File

@@ -1 +1,8 @@
.private/
.venv/
.env
content/
assets/
wikijs_id_ed25519*
__pycache__/
*.pyc

1
.python-version Normal file
View File

@@ -0,0 +1 @@
3.13

0
assets/images/.gitkeep → README.md
View File

39
content/ships/caldari/frigates/condor.md
View File

@@ -1,39 +0,0 @@
---
title: Condor
source: eve-university
---
Summary
The Condor is the Caldari Attack Frigate. The Condor's speed, agility, and bonus to propulsion jamming modules (warp disruptors, warp scramblers and stasis webifiers) make it a good tackle frigate. The ship’s four mid slots allow it to be fit with more than one propulsion jammer, or a variety of other support modules, making the Condor tricky to predict and counter. The Condor also has bonuses to light missile and rocket damage. In principle, all of this makes the Condor a versatile Attack Frigate, capable of being flown in a number of different ways. In practice, however, the Condor is relatively slow and difficult to fit, especially for low-skill pilots. New, Caldari-focused players that want to fly fast tackle have a tough choice between piloting the Condor in spite of its limitations, or cross-training into a different racial tackler.
The Condor’s versatility makes it a good platform for new pilots learning different approaches to tackling and scouting. Flying the Condor provides practice for eventually piloting fleet interceptors (such as the Crow), which are more commonly used as fleet scouts and first tacklers. Nevertheless, even though the Tech I Condor isn't quite as effective as a Tech II Interceptor, it can still serve as a good tackle ship. E-UNI Condor pilots are strongly advised to read the UniWiki guide to tackling.
The Condor has the lowest base HP of all Attack Frigates, much of which is focused on shield. As a result, the Condor is typically fit with a mid-slot shield extender and shield rigs. The tackle Condor is also typically equipped with a microwarpdrive and a warp disruptor (“point”) in two of its mid slots, like most fast tackle frigates. The fourth mid slot can be fit with a variety of different modules—a warp scrambler (“scram”), a stasis webifier (“web”), a sensor booster for fast lock time, or even a tracking or guidance disruptor to mitigate enemy damage. Different fits require different tactics, some of which are described in the Tactics section below. In the end, Condor pilots are encouraged to try out different layouts, and experience first-hand how their choices affect the way the ship flies.
The Condor has the least powergrid (PG) of the Attack Frigates, and that means that once the ship’s mid slots have been filled, there is little room for anything else. The Condor’s bonuses to missiles and rockets are difficult to take advantage of, because launchers require PG that the Condor doesn’t have. Tackle Condors can instead be fit with no weapons at all. The Condor can also be fit as a kiting missile or rocket frigate, but only by sacrificing tank. Caldari characters looking for a PvE frigate, or a “second” (scram/web) tackler, would be much better off flying the Merlin Combat Frigate.
Skills
Missile skills - these will increase the damage and range of missiles/rockets. These are more important when flying solo or in a damage dealer than when flying tackle.
Shield skills - these will make it easier to fit shield extenders (Shield Upgrades in particular) and increases their effectiveness. Helpful for all roles.
Weapon Disruption skills - these will help only if flying with weapons disruptors fit which are present on some tackle fits.
Tactics
Fleet Condor pilots should read the UniWiki articles on Scouting and Tackling; these provide details on how to serve as the fleet vanguard, and how to approach and point a target once it is on grid. Some tips on flying the tackle Condor:
Choose your targets wisely. The Condor is especially effective against large ships, because they have trouble tracking the Condor with their larger weapons systems. Condor pilots can make a bid on any ship that is battlecruiser-sized and larger, on most cruisers, and even some destroyers. Anti-tackle faction cruisers, Tech III destroyers, and enemy frigates, on the other hand, will still outrun the rest of your fleet if they are fit for speed, even if you stay with and point them. If you have fitted a scram or web in addition to your point, you can tackle these speedier targets. But, be prepared to take some damage--ships like these are typically fit to hit tackle frigates like yours.
Keep an appropriate distance. If you are only fit with a warp disruptor, check its optimal range, and subtract 4 km from that number; this is the distance at which you’ll want to orbit your target. (This gives you 4 km of wiggle room.) If you stray too far below this distance, you will be within range of your opponent’s shorter-range scram, web and/or neuts. You can set your ship’s custom default orbit by right-clicking on the Orbit button in the Selected Item window; you should set your default 2 km less than the number you calculated above, because of overshoot at high speeds.
If you are fit with a scram, and you want to use it on your target, set your orbit to 500 meters and shut off your microwarpdrive (MWD) once you get within ~10 km of your target. This will allow your ship to quickly achieve a stable orbit at extremely close range to hopefully get under and stay under enemy guns. If you turn your MWD off too late you will 'yo-yo' on the target and potentially lose your scram. Leaving your MWD on drains capacitor unnecessarily, makes it much easier for other ships to shoot you (due to the signature bloom of an active MWD), and may force you into an elliptical orbit that pushes you outside of scram range.
Spiral in. Because the Condor is poorly tanked, the Condor pilot needs to avoid getting hit on approach. If the target is a turret ship, one way to avoid damage is to "spiral" towards the target, and to never fly directly at the target, in order to maximize transversal (and thus evade the tracking of the target's guns). The spiraling method does not help against opponents flying missile or drone ships, though, and Condor pilots should just fly straight at these targets. The Manual Piloting page has more details.
Anticipate your opponent. Try and determine your opponent’s direction of flight, and cut them off (by double clicking in front of them). Your opponent may try to "slingshot" you by flying towards you, and making you overshoot the target. Be ready to change directions quickly.
Call your point in comms. Let your fleetmates know when they should warp to you!
Notes
Condors are one of two types of birds, named after the places where they live; the California Condor and the Andean Condor.
Category:Ship Database
Category:Standard Frigates

6
main.py Normal file
View File

@@ -0,0 +1,6 @@
def main():
print("Hello from eve-wiki-agents!")
if __name__ == "__main__":
main()

21
pyproject.toml Normal file
View File

@@ -0,0 +1,21 @@
[project]
name = "eve-wiki-agents"
version = "0.1.0"
description = "Add your description here"
readme = "README.md"
requires-python = ">=3.13"
dependencies = [
"aiolimiter>=1.2.1",
"beautifulsoup4>=4.14.3",
"feedparser>=6.0.12",
"gitpython>=3.1.46",
"httpx>=0.28.1",
"langchain>=1.2.15",
"langchain-openai>=1.1.14",
"langgraph>=1.1.8",
"mwparserfromhell>=0.7.2",
"psycopg>=3.3.3",
"pydantic>=2.13.2",
"python-dotenv>=1.2.2",
"redis>=7.4.0",
]

0
content/.gitkeep → src/agents/__init__.py
View File

69
src/agents/esi_collector.py Normal file
View File

@@ -0,0 +1,69 @@
from typing import Dict, Any
from src.schema.state import WikiState, ShipData
from src.clients.esi import ESIClient
import logging
logger = logging.getLogger("esi-collector")
async def get_dogma_attribute(esi: ESIClient, type_id: int, attribute_id: int) -> float:
"""Helper to get a specific dogma attribute value for a type."""
type_details = await esi.get_type_details(type_id)
for attr in type_details.get("dogma_attributes", []):
if attr.get("attribute_id") == attribute_id:
return float(attr.get("value", 0))
return 0.0
async def esi_collector_node(state: WikiState) -> Dict[str, Any]:
"""
LangGraph node to collect structured data from ESI.
"""
esi = ESIClient()
# Common Attribute IDs (ESI Dogma)
ATTR_HP_SHIELD = 263
ATTR_HP_ARMOR = 265
ATTR_HP_STRUCT = 9
ATTR_CPU_OUTPUT = 48
ATTR_PG_OUTPUT = 11
try:
if state.current_page and state.current_page.metadata.page_type == "ship":
type_id = state.current_page.frontmatter.get("type_id")
if type_id:
logger.info(f"Collecting ESI data for type_id {type_id}")
type_details = await esi.get_type_details(type_id)
# Fetch group for category verification
group = await esi.get_group_details(type_details.get("group_id"))
# Map attributes
dogma = {a["attribute_id"]: a["value"] for a in type_details.get("dogma_attributes", [])}
ship_data = ShipData(
type_id=type_id,
group=group.get("name", "Unknown"),
race="Unknown", # Need race mapping
hull_stats={
"hp_shield": int(dogma.get(ATTR_HP_SHIELD, 0)),
"hp_armor": int(dogma.get(ATTR_HP_ARMOR, 0)),
"hp_structure": int(dogma.get(ATTR_HP_STRUCT, 0))
},
fitting_stats={
"cpu_output": int(dogma.get(ATTR_CPU_OUTPUT, 0)),
"powergrid_output": int(dogma.get(ATTR_PG_OUTPUT, 0))
},
velocity=int(dogma.get(37, 0)), # Base velocity attribute
skill_requirements=[]
)
logger.info(f"Successfully collected data for {type_details.get('name')}")
# We update the state with the new data overlay
return {"current_page": state.current_page} # State update logic needed
return {}
except Exception as e:
logger.error(f"ESI Collection Failed: {str(e)}")
return {"error": str(e)}
finally:
await esi.close()

70
src/agents/harvester.py Normal file
View File

@@ -0,0 +1,70 @@
from typing import Dict, Any
from src.schema.state import WikiState, WikiPage, PageMetadata
from src.clients.mediawiki import MediaWikiClient
from src.utils.converter import wikitext_to_markdown
from datetime import datetime
import logging
logger = logging.getLogger("harvester-agent")
async def harvester_node(state: WikiState) -> Dict[str, Any]:
"""
Agent A: Source Harvester.
Extracts content from MediaWiki and converts it to Markdown.
"""
mw = MediaWikiClient()
try:
# Example: If we're starting a new run and have no current page
# In a real run, this would be driven by a list of pages to import
if not state.current_page:
title = "Condor" # Hardcoded for seeding/test
logger.info(f"Harvesting content for page: {title}")
wikitext = await mw.get_page_wikitext(title)
if not wikitext:
return {"error": f"Page {title} not found on source wiki."}
logger.info("Converting wikitext to markdown...")
try:
markdown = wikitext_to_markdown(wikitext)
except Exception as conv_err:
logger.error(f"Conversion Failed: {str(conv_err)}")
raise
logger.info("Creating WikiPage object...")
# Create the initial WikiPage object
metadata = PageMetadata(
page_type="ship",
source="eve-university",
source_url=f"https://wiki.eveuniversity.org/{title}",
imported_date=datetime.now(),
last_updated=datetime.now(),
last_validated=datetime.now(),
update_frequency="weekly",
validation_score=0.0,
categories=["ships", "frigates"]
)
try:
new_page = WikiPage(
path=f"ships/caldari/frigates/{title.lower()}",
title=title,
content_markdown=markdown,
metadata=metadata,
frontmatter={"type_id": 582}, # Known ID for Condor
assets=[]
)
except Exception as pydantic_err:
logger.error(f"Pydantic Validation Failed: {str(pydantic_err)}")
raise
return {"current_page": new_page}
return {}
except Exception as e:
logger.error(f"Harvesting Failed: {str(e)}")
return {"error": str(e)}
finally:
await mw.close()

68
src/agents/validation.py Normal file
View File

@@ -0,0 +1,68 @@
from typing import Dict, Any, List
from src.schema.state import WikiState, ValidationResult
import logging
logger = logging.getLogger("validation-agent")
async def validation_agent_node(state: WikiState) -> Dict[str, Any]:
"""
Agent E & F: Combined Validation Layer.
Calculates weighted confidence score and enforces 'Must Pass' rules.
"""
results: List[ValidationResult] = []
# 1. Numerical Validation (Agent F) - 40%
# In a real run, this compares state.current_page against ESI
esi_passed = True # Placeholder
results.append(ValidationResult(
category="numerical",
passed=esi_passed,
score=1.0 if esi_passed else 0.0,
feedback=None if esi_passed else "ESI Data mismatch detected."
))
# 2. Structural Validation - 20%
struct_passed = True # Placeholder
results.append(ValidationResult(
category="structural",
passed=struct_passed,
score=1.0 if struct_passed else 0.0,
feedback=None if struct_passed else "Markdown template violation."
))
# 3. Relational Validation - 20%
relat_passed = True
results.append(ValidationResult(
category="relational",
passed=relat_passed,
score=1.0,
feedback=None
))
# 4. Semantic Validation - 20%
semant_passed = True
results.append(ValidationResult(
category="semantic",
passed=semant_passed,
score=1.0,
feedback=None
))
# Calculate Weighted Score
# Total Score = (ESI * 0.4) + (Struct * 0.2) + (Relat * 0.2) + (Semant * 0.2)
total_score = sum(r.score * weight for r, weight in zip(results, [0.4, 0.2, 0.2, 0.2]))
# Enforce "Must Pass" Rule
must_pass_failed = any(not r.passed for r in results if r.category in ["numerical", "structural"])
if must_pass_failed:
logger.warning("Must-Pass validation category failed! Capping score at 0.")
total_score = 0.0
is_approved = total_score >= 0.95
logger.info(f"Validation Complete. Score: {total_score:.2f}. Approved: {is_approved}")
return {
"validation_pipeline_results": results,
"is_approved": is_approved
}

0
src/clients/__init__.py Normal file
View File

69
src/clients/esi.py Normal file
View File

@@ -0,0 +1,69 @@
import httpx
import asyncio
from aiolimiter import AsyncLimiter
from typing import List, Dict, Any, Optional
import logging
# Configure logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger("esi-client")
class ESIClient:
BASE_URL = "https://esi.evetech.net/latest"
def __init__(self, user_agent: str = "EVE-Wiki-Agent/1.0", rate_limit: int = 20):
self.client = httpx.AsyncClient(
base_url=self.BASE_URL,
headers={"User-Agent": user_agent},
timeout=30.0
)
# CCP recommends 20 req/s, but we'll use a limiter to be safe
self.limiter = AsyncLimiter(rate_limit, 1)
async def close(self):
await self.client.aclose()
async def _request(self, method: str, path: str, params: Optional[Dict] = None) -> Any:
async with self.limiter:
response = await self.client.request(method, path, params=params)
# Handle rate limit headers from ESI if they exist
if response.status_code == 429:
logger.warning("ESI Rate Limit Hit! Sleeping...")
await asyncio.sleep(5)
return await self._request(method, path, params)
if response.status_code == 404:
return None
response.raise_for_status()
return response.json()
async def get_universe_groups(self) -> List[int]:
"""Fetch all universe groups."""
return await self._request("GET", "/universe/groups/")
async def get_group_details(self, group_id: int) -> Dict[str, Any]:
"""Fetch details for a specific group."""
return await self._request("GET", f"/universe/groups/{group_id}/")
async def get_type_details(self, type_id: int) -> Dict[str, Any]:
"""Fetch details for a specific TypeID."""
return await self._request("GET", f"/universe/types/{type_id}/")
async def get_category_details(self, category_id: int) -> Dict[str, Any]:
"""Fetch details for a specific category."""
return await self._request("GET", f"/universe/categories/{category_id}/")
async def get_all_ship_types(self) -> List[int]:
"""Helper to get all ship type IDs (Category 6)."""
ship_category = await self.get_category_details(6)
all_types = []
# This is expensive as it requires crawling groups
# In a real run, we'd cache these or use a static map
for group_id in ship_category.get("groups", []):
group = await self.get_group_details(group_id)
all_types.extend(group.get("types", []))
return all_types

64
src/clients/mediawiki.py Normal file
View File

@@ -0,0 +1,64 @@
import httpx
import asyncio
from aiolimiter import AsyncLimiter
from typing import List, Dict, Any, Optional
import logging
logger = logging.getLogger("mediawiki-client")
class MediaWikiClient:
def __init__(self, api_url: str = "https://wiki.eveuniversity.org/api.php", rate_limit: int = 2):
self.api_url = api_url
self.client = httpx.AsyncClient(
timeout=30.0,
headers={"User-Agent": "EVE-Wiki-Agent/1.0 (admin@ben.io)"}
)
self.limiter = AsyncLimiter(rate_limit, 1)
async def close(self):
await self.client.aclose()
async def _request(self, params: Dict[str, Any]) -> Any:
# Default params
params.setdefault("format", "json")
params.setdefault("action", "query")
async with self.limiter:
response = await self.client.get(self.api_url, params=params)
response.raise_for_status()
return response.json()
async def get_page_wikitext(self, title: str) -> Optional[str]:
"""Fetch the raw wikitext of a page."""
params = {
"prop": "revisions",
"titles": title,
"rvslots": "*",
"rvprop": "content"
}
data = await self._request(params)
pages = data.get("query", {}).get("pages", {})
for page_id, page_data in pages.items():
if page_id == "-1":
return None
revisions = page_data.get("revisions", [])
if revisions:
return revisions[0].get("slots", {}).get("main", {}).get("*")
return None
async def get_category_members(self, category: str, cmtype: str = "page") -> List[str]:
"""List all members of a category."""
params = {
"list": "categorymembers",
"cmtitle": f"Category:{category}",
"cmtype": cmtype,
"cmlimit": 500
}
data = await self._request(params)
members = data.get("query", {}).get("categorymembers", [])
return [m.get("title") for m in members]
async def get_all_ships(self) -> List[str]:
"""Helper to get all ship page titles."""
return await self.get_category_members("Ships")

54
src/clients/rss.py Normal file
View File

@@ -0,0 +1,54 @@
import httpx
import feedparser
from typing import List, Dict, Any
import logging
logger = logging.getLogger("rss-client")
class RSSClient:
RSS_URL = "https://www.eveonline.com/rss/patch-notes"
def __init__(self, user_agent: str = "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/123.0.0.0 Safari/537.36"):
self.client = httpx.AsyncClient(
headers={"User-Agent": user_agent},
follow_redirects=True,
timeout=30.0
)
async def close(self):
await self.client.aclose()
async def get_latest_patch_notes(self) -> List[Dict[str, Any]]:
"""Fetch and parse the latest patch notes."""
try:
response = await self.client.get(self.RSS_URL)
response.raise_for_status()
feed = feedparser.parse(response.text)
entries = []
for entry in feed.entries:
entries.append({
"title": entry.title,
"link": entry.link,
"published": entry.published if hasattr(entry, "published") else None,
"summary": entry.description if hasattr(entry, "summary") else entry.get("description", "")
})
return entries
except Exception as e:
logger.error(f"RSS Fetch Failed: {str(e)}")
return []
if __name__ == "__main__":
import asyncio
async def test():
rss = RSSClient()
notes = await rss.get_latest_patch_notes()
print(f"Found {len(notes)} patch note entries.")
if notes:
print(f"Latest: {notes[0]['title']} ({notes[0]['link']})")
await rss.close()
asyncio.run(test())

58
src/clients/wckg.py Normal file
View File

@@ -0,0 +1,58 @@
import httpx
from bs4 import BeautifulSoup
from typing import Optional, Dict, Any
import logging
logger = logging.getLogger("wckg-client")
class WCKGClient:
BASE_URL = "https://www.wckg.net"
def __init__(self, user_agent: str = "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/123.0.0.0 Safari/537.36"):
self.client = httpx.AsyncClient(
base_url=self.BASE_URL,
headers={"User-Agent": user_agent},
follow_redirects=True,
timeout=30.0
)
async def close(self):
await self.client.aclose()
async def get_page_content(self, path: str) -> Optional[str]:
"""Fetch and extract content from a WCKG Google Sites page."""
try:
response = await self.client.get(path)
response.raise_for_status()
soup = BeautifulSoup(response.text, "html.parser")
# Google Sites specific content selector
# Usually the main content is within a 'section' or has specific data-attr
content_area = soup.find("section") or soup.find("div", {"role": "main"})
if not content_area:
# Fallback to body if selector fails
content_area = soup.body
return content_area.get_text(separator="\n", strip=True)
except Exception as e:
logger.error(f"WCKG Fetch Failed for {path}: {str(e)}")
return None
if __name__ == "__main__":
import asyncio
async def test():
wckg = WCKGClient()
# Test fetching a known guide
print("Testing WCKG: Fetching Exploration guide...")
content = await wckg.get_page_content("/pve/exploration")
if content:
print(f"Success: Fetched {len(content)} characters.")
print(f"Preview: {content[:100]}...")
else:
print("Failed: Content not found.")
await wckg.close()
asyncio.run(test())

125
src/clients/wikijs.py Normal file
View File

@@ -0,0 +1,125 @@
import httpx
import os
from typing import List, Dict, Any, Optional
import logging
from dotenv import load_dotenv
load_dotenv()
logger = logging.getLogger("wikijs-client")
class WikiJSClient:
def __init__(self):
self.api_url = "https://eve-wiki.ext.ben.io/graphql"
self.api_token = os.getenv("WIKI_API_KEY")
if not self.api_token:
logger.warning("WIKI_API_KEY not found in environment!")
self.client = httpx.AsyncClient(
headers={
"Authorization": f"Bearer {self.api_token}",
"Content-Type": "application/json"
},
timeout=30.0
)
async def close(self):
await self.client.aclose()
async def execute_graphql(self, query: str, variables: Optional[Dict] = None) -> Any:
payload = {"query": query}
if variables:
payload["variables"] = variables
response = await self.client.post(self.api_url, json=payload)
response.raise_for_status()
data = response.json()
# Don't raise if it's just a 'Not Found' error for singleByPath
if "errors" in data:
if any(err.get("message") == "This page does not exist." for err in data["errors"]):
return {"pages": {"singleByPath": None}}
logger.error(f"GraphQL Errors: {data['errors']}")
raise Exception(f"Wiki.js API Error: {data['errors'][0]['message']}")
return data.get("data")
async def get_page_by_path(self, path: str) -> Optional[Dict]:
"""Fetch a page by its path."""
query = """
query($path: String!) {
pages {
singleByPath(path: $path, locale: "en") {
id, title, content, description
}
}
}
"""
data = await self.execute_graphql(query, {"path": path})
return data.get("pages", {}).get("singleByPath")
async def upsert_page(self, path: str, title: str, content: str, description: str = "", tags: List[str] = []) -> Dict:
"""Create or update a page."""
existing = await self.get_page_by_path(path)
if existing:
# Update
query = """
mutation($id: Int!, $title: String!, $content: String!, $description: String!, $tags: [String]!) {
pages {
update(id: $id, title: $title, content: $content, description: $description, tags: $tags) {
responseResult { succeeded, message }
}
}
}
"""
variables = {
"id": int(existing["id"]),
"title": title,
"content": content,
"description": description,
"tags": tags
}
else:
# Create
query = """
mutation($title: String!, $content: String!, $description: String!, $path: String!, $tags: [String]!, $isPublished: Boolean!, $isPrivate: Boolean!) {
pages {
create(title: $title, content: $content, description: $description, path: $path, tags: $tags, editor: "markdown", locale: "en", isPublished: $isPublished, isPrivate: $isPrivate) {
responseResult { succeeded, message }
}
}
}
"""
variables = {
"title": title,
"content": content,
"description": description,
"path": path,
"tags": tags,
"isPublished": True,
"isPrivate": False
}
return await self.execute_graphql(query, variables)
if __name__ == "__main__":
import asyncio
async def test():
wiki = WikiJSClient()
try:
print("Testing Wiki.js: Creating test page...")
res = await wiki.upsert_page(
path="test/agent-connection",
title="Agent Connection Test",
content="# Success\nThis page was created by the automation agent.",
description="Automated API test",
tags=["test", "agent"]
)
print(f"Result: {res}")
finally:
await wiki.close()
asyncio.run(test())

132
src/main.py Normal file
View File

@@ -0,0 +1,132 @@
from langgraph.graph import StateGraph, END
from src.schema.state import WikiState
from src.agents.harvester import harvester_node
from src.agents.esi_collector import esi_collector_node
from src.agents.validation import validation_agent_node
from typing import Dict, Any
# --- Nodes ---
import git
import os
async def git_sync_node(state: WikiState) -> Dict[str, Any]:
"""Infrastructure: Sync content to Git SSOT."""
print("Node: Git Sync")
if not state.current_page:
return {}
repo_path = os.getcwd()
repo = git.Repo(repo_path)
# Ensure directory exists
file_path = f"content/{state.current_page.path}.md"
os.makedirs(os.path.dirname(file_path), exist_ok=True)
# Write content
with open(file_path, "w") as f:
# Write frontmatter (simplified)
f.write("---\n")
f.write(f"title: {state.current_page.title}\n")
f.write(f"source: {state.current_page.metadata.source}\n")
f.write("---\n\n")
f.write(state.current_page.content_markdown)
# Commit and Push
try:
repo.index.add([file_path])
repo.index.commit(f"[AGENT] Update: {state.current_page.path}")
repo.remotes.origin.push()
print(f"Successfully pushed {file_path} to Git.")
except Exception as e:
print(f"Git Sync Failed: {str(e)}")
return {}
from src.clients.wikijs import WikiJSClient
async def wikijs_node(state: WikiState) -> Dict[str, Any]:
"""Agent API: Publish to Wiki.js UI."""
print("Node: Wiki.js API")
if not state.current_page:
return {}
wiki = WikiJSClient()
try:
res = await wiki.upsert_page(
path=state.current_page.path,
title=state.current_page.title,
content=state.current_page.content_markdown,
description=f"Automated sync from {state.current_page.metadata.source}",
tags=state.current_page.metadata.categories
)
print(f"Successfully published to Wiki.js: {state.current_page.path}")
except Exception as e:
print(f"Wiki.js Publication Failed: {str(e)}")
finally:
await wiki.close()
return {}
# --- Routing ---
def should_publish(state: WikiState) -> str:
"""Route to Git/Wiki or Halt based on approval."""
if state.is_approved:
return "publish"
return "halt"
# --- Graph Construction ---
def create_wiki_graph():
workflow = StateGraph(WikiState)
# Add Nodes
workflow.add_node("harvester", harvester_node)
workflow.add_node("esi_collector", esi_collector_node)
workflow.add_node("validation", validation_agent_node)
workflow.add_node("git_sync", git_sync_node)
workflow.add_node("wikijs_api", wikijs_node)
# Define Edges
workflow.set_entry_point("harvester")
workflow.add_edge("harvester", "esi_collector")
workflow.add_edge("esi_collector", "validation")
# Conditional Routing
workflow.add_conditional_edges(
"validation",
should_publish,
{
"publish": "git_sync",
"halt": END
}
)
workflow.add_edge("git_sync", "wikijs_api")
workflow.add_edge("wikijs_api", END)
return workflow.compile()
if __name__ == "__main__":
import asyncio
async def run_test():
graph = create_wiki_graph()
print("--- Starting Wiki Automation Test (Condor) ---")
initial_state = {
"thread_id": "test-run-1",
"retry_count": 0,
"is_approved": False
}
async for event in graph.astream(initial_state):
for node_name, output in event.items():
print(f"\n[Node: {node_name}]")
if output and isinstance(output, dict) and "error" in output:
print(f"Error: {output['error']}")
print("\n--- Test Run Complete ---")
asyncio.run(run_test())

63
src/schema/state.py Normal file
View File

@@ -0,0 +1,63 @@
from typing import List, Optional, Dict, Annotated
from pydantic import BaseModel, Field
from datetime import datetime
class ContentSource(BaseModel):
name: str # e.g., "eve-university", "wckg", "esi"
url: str
content_hash: str
extracted_at: datetime
class ValidationResult(BaseModel):
category: str # "structural", "content", "numerical", "cross-reference"
passed: bool
score: float # 0.0 to 1.0
feedback: Optional[str] = None
details: Dict = {}
class PageMetadata(BaseModel):
page_type: str # "ship", "module", "mechanic", "guide"
source: str
source_url: str
imported_date: datetime
last_updated: datetime
last_validated: datetime
update_frequency: str
validation_score: float
categories: List[str]
class WikiPage(BaseModel):
path: str # e.g., "ships/frigates/condor"
title: str
content_markdown: str
metadata: PageMetadata
frontmatter: Dict
assets: List[str] # List of local asset paths
class WikiState(BaseModel):
# Core processing data
current_page: Optional[WikiPage] = None
proposed_changes: List[WikiPage] = []
# Pipeline tracking
sources: List[ContentSource] = []
validation_pipeline_results: List[ValidationResult] = []
# Control flow
retry_count: int = 0
max_retries: int = 3
is_approved: bool = False
error: Optional[str] = None
# Checkpointing metadata
thread_id: str
state_checkpoint_id: Optional[str] = None
class ShipData(BaseModel):
type_id: int
group: str
race: str
hull_stats: Dict[str, int]
fitting_stats: Dict[str, int]
velocity: int
skill_requirements: List[Dict[str, int]]

0
src/utils/__init__.py Normal file
View File

43
src/utils/converter.py Normal file
View File

@@ -0,0 +1,43 @@
import mwparserfromhell
import re
from typing import Optional
def wikitext_to_markdown(wikitext: str) -> str:
"""
Converts MediaWiki wikitext to basic Markdown.
"""
if not wikitext:
return ""
# Parse wikitext
parsed = mwparserfromhell.parse(wikitext)
# 1. Handle Templates
# We use a list to avoid modifying while iterating
for template in list(parsed.filter_templates()):
try:
parsed.remove(template)
except ValueError:
pass # Already removed
# 2. Get cleaned text
text = str(parsed.strip_code())
# 3. Basic Markdown Regex Fixes
# Convert Headers (== Header == -> ## Header)
# strip_code handles some of this, but we want to ensure markdown compatibility
# Convert Bold/Italic
text = re.sub(r"'''(.*?)'''", r"**\1**", text)
text = re.sub(r"''(.*?)''", r"*\1*", text)
# Clean up double newlines
text = re.sub(r"\n{3,}", "\n\n", text)
return text.strip()
if __name__ == "__main__":
# Quick Test
sample = "== Condor ==\nThe '''Condor''' is a [[Caldari]] frigate.\n* Fast\n* Cheap"
print(wikitext_to_markdown(sample))

24
tests/test_esi.py Normal file
View File

@@ -0,0 +1,24 @@
import asyncio
from src.clients.esi import ESIClient
import logging
async def test_esi():
esi = ESIClient()
try:
# Test 1: Fetch a known TypeID (e.g., Condor = 603)
print("Testing ESI: Fetching Condor (TypeID 603)...")
condor = await esi.get_type_details(603)
print(f"Success: Found {condor.get('name')}")
# Test 2: Rate Limit Test (Fetch 5 types rapidly)
print("\nTesting Rate Limiter: Fetching 5 items...")
tasks = [esi.get_type_details(603 + i) for i in range(5)]
results = await asyncio.gather(*tasks)
for res in results:
print(f" - Fetched: {res.get('name')}")
finally:
await esi.close()
if __name__ == "__main__":
asyncio.run(test_esi())

28
tests/test_mediawiki.py Normal file
View File

@@ -0,0 +1,28 @@
import asyncio
from src.clients.mediawiki import MediaWikiClient
import logging
async def test_mediawiki():
mw = MediaWikiClient()
try:
# Test 1: Fetch a known page (e.g., "Condor")
print("Testing MediaWiki: Fetching Condor page...")
content = await mw.get_page_wikitext("Condor")
if content:
print(f"Success: Fetched {len(content)} characters of wikitext.")
print(f"Preview: {content[:100]}...")
else:
print("Failed: Page not found.")
# Test 2: List members of a category
print("\nTesting Category Members: Fetching 'Frigates'...")
members = await mw.get_category_members("Frigates")
print(f"Success: Found {len(members)} pages in Category:Frigates.")
for member in members[:5]:
print(f" - Page: {member}")
finally:
await mw.close()
if __name__ == "__main__":
asyncio.run(test_mediawiki())

1058
uv.lock generated Normal file
View File

File diff suppressed because it is too large Load Diff