MCPcopy Create free account
hub / github.com/RASAAS/docmcp-knowledge / DatabaseComparator

Class DatabaseComparator

scripts/fetch_updates.py:458–611  ·  view source on GitHub ↗

Loads all _index.json files and version_registry.json to classify detections.

Source from the content-addressed store, hash-verified

456# ---------------------------------------------------------------------------
457
458class DatabaseComparator:
459 """Loads all _index.json files and version_registry.json to classify detections."""
460
461 def __init__(self, root: Path = ROOT):
462 self.root = root
463 self.mdcg_ids: set[str] = set()
464 self.mdcg_titles: set[str] = set()
465 self.standards_latest_amendment: str = ""
466 self.eu_reg_ids: set[str] = set()
467 self.nmpa_reg_titles: set[str] = set()
468 self.nmpa_guidance_titles: set[str] = set()
469 self.fda_guidance_titles: set[str] = set()
470 self.fda_guidance_ids: set[str] = set()
471 self.db_stats: dict[str, dict] = {}
472 self._load_all()
473
474 def _load_json(self, path: Path) -> dict:
475 if not path.exists():
476 return {}
477 try:
478 with open(path, "r", encoding="utf-8") as f:
479 return json.load(f)
480 except Exception:
481 return {}
482
483 def _load_all(self):
484 mdcg = self._load_json(self.root / "eu_mdr" / "mdcg" / "_index.json")
485 for entry in mdcg.get("entries", []):
486 eid = entry.get("id", "")
487 self.mdcg_ids.add(eid)
488 for lang in ("zh", "en"):
489 t = (entry.get("title") or {}).get(lang, "")
490 if t:
491 self.mdcg_titles.add(t.strip().lower())
492 self._mdcg_entries = mdcg.get("entries", [])
493 self.db_stats["eu_mdr/mdcg"] = {"count": len(self.mdcg_ids)}
494
495 stds = self._load_json(self.root / "eu_mdr" / "standards" / "_index.json")
496 self.standards_latest_amendment = stds.get("latest_amendment", "")
497 self.db_stats["eu_mdr/standards"] = {
498 "count": stds.get("total_standards", 0),
499 "latest_amendment": self.standards_latest_amendment,
500 }
501
502 regs = self._load_json(self.root / "eu_mdr" / "regulations" / "_index.json")
503 for doc in regs.get("documents", []):
504 self.eu_reg_ids.add(doc.get("id", ""))
505 self.db_stats["eu_mdr/regulations"] = {"count": len(self.eu_reg_ids)}
506
507 nmpa_reg = self._load_json(self.root / "nmpa" / "regulations" / "_index.json")
508 for entry in nmpa_reg.get("entries", []):
509 t = (entry.get("title") or {}).get("zh", "")
510 if t:
511 self.nmpa_reg_titles.add(t.strip())
512 self.db_stats["nmpa/regulations"] = {"count": len(self.nmpa_reg_titles)}
513
514 nmpa_guide = self._load_json(self.root / "nmpa" / "guidance" / "_index.json")
515 for entry in nmpa_guide.get("entries", []):

Callers 1

__init__Method · 0.85

Calls

no outgoing calls

Tested by

no test coverage detected