m2 (AI Agent)
1d10589cef
Registry (Forgejo) + memory-api clients under m2_market_indexer.client, with memory-api paths/creds as env-overridable constants (config.py). reindex.py drops market:catalog and rebuilds from published listings on main, skipping schema-invalid ones. watch.py polls the registry tip commit against a cached state.json and upserts changed/new published listings, deletes delisted ones.
93 lines
2.8 KiB
Python
93 lines
2.8 KiB
Python
"""Full rebuild of market:catalog from the registry `main` (contracts/catalog-index.md).
|
|
|
|
Drop partition, then upsert every `published` listing found under `listings/`
|
|
on the registry's default branch. Invalid listing.json documents are skipped
|
|
with a warning, not fatal — one bad listing must not block the rebuild
|
|
(constitution VII: smallest thing that closes the loop).
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import sys
|
|
|
|
import httpx
|
|
from jsonschema import Draft202012Validator
|
|
|
|
from .client import MemoryClient, RegistryClient, build_embedded_text
|
|
from .config import listing_schema_path, load_config
|
|
|
|
|
|
def _load_validator() -> Draft202012Validator:
|
|
import json
|
|
|
|
with listing_schema_path().open() as f:
|
|
schema = json.load(f)
|
|
return Draft202012Validator(schema)
|
|
|
|
|
|
def run(registry: RegistryClient, memory: MemoryClient, validator: Draft202012Validator) -> dict:
|
|
"""Execute one full reindex; returns counts for reporting."""
|
|
counts = {"scanned": 0, "invalid": 0, "not_published": 0, "upserted": 0}
|
|
|
|
listing_ids = registry.list_listing_ids()
|
|
memory.drop_partition()
|
|
|
|
for listing_id in listing_ids:
|
|
counts["scanned"] += 1
|
|
try:
|
|
listing = registry.get_listing(listing_id)
|
|
except (httpx.HTTPStatusError, ValueError) as exc:
|
|
print(f"WARN: skipping {listing_id}: cannot read listing.json ({exc})", file=sys.stderr)
|
|
counts["invalid"] += 1
|
|
continue
|
|
|
|
errors = list(validator.iter_errors(listing))
|
|
if errors:
|
|
print(
|
|
f"WARN: skipping {listing_id}: schema violations: "
|
|
+ "; ".join(e.message for e in errors),
|
|
file=sys.stderr,
|
|
)
|
|
counts["invalid"] += 1
|
|
continue
|
|
|
|
if listing.get("status") != "published":
|
|
counts["not_published"] += 1
|
|
continue
|
|
|
|
solution = registry.get_solution(listing_id)
|
|
text = build_embedded_text(listing, solution)
|
|
memory.upsert_document(listing_id, text, listing)
|
|
counts["upserted"] += 1
|
|
|
|
return counts
|
|
|
|
|
|
def main() -> None:
|
|
config = load_config()
|
|
validator = _load_validator()
|
|
|
|
with RegistryClient(
|
|
config.forgejo_url,
|
|
config.forgejo_token,
|
|
config.registry_repo,
|
|
config.registry_ref,
|
|
) as registry, MemoryClient(
|
|
config.memory_api_url,
|
|
config.memory_api_key,
|
|
config.memory_api_partition,
|
|
partition_path=config.memory_api_partition_path,
|
|
document_path=config.memory_api_document_path,
|
|
) as memory:
|
|
counts = run(registry, memory, validator)
|
|
|
|
print(
|
|
f"reindex complete: scanned={counts['scanned']} "
|
|
f"upserted={counts['upserted']} "
|
|
f"not_published={counts['not_published']} "
|
|
f"invalid_skipped={counts['invalid']}"
|
|
)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|