Spaces:
Build error
Build error
File size: 3,495 Bytes
9f48a3d 8a0fba4 9f48a3d | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 | import os, sys, json
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__)))))
from datetime import datetime
from loguru import logger
KNOWN_POLITICIAN_FIELDS = {
"id","name","state","party","constituency","criminal_cases",
"total_assets_crore","movable_assets_crore","education","year",
"risk_score","risk_level","betweenness_centrality","pagerank",
}
KNOWN_COMPANY_FIELDS = {
"id","name","state","cin","status","paid_up_capital_crore",
"industry","registration_date","director_count",
}
KNOWN_CONTRACT_FIELDS = {
"id","order_id","item_desc","amount_crore","buyer_org",
"order_date","company_id","ministry","category",
}
PENDING_FILE = os.path.join(
os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__)))),
"data", "processed", "pending_schema_additions.json"
)
NODE_SCHEMAS = {
"Politician": KNOWN_POLITICIAN_FIELDS,
"Company": KNOWN_COMPANY_FIELDS,
"Contract": KNOWN_CONTRACT_FIELDS,
}
class SchemaLearner:
def detect_new_fields(self, records: list[dict],
node_type: str) -> dict:
known = NODE_SCHEMAS.get(node_type, set())
new_found = {}
for record in records:
for field, value in record.items():
if field not in known and field not in new_found:
new_found[field] = {
"sample_value": str(value)[:100],
"node_type": node_type,
"detected_at": datetime.now().isoformat(),
"status": "pending_review",
}
if new_found:
logger.info(
f"[SchemaLearner] {len(new_found)} new fields in {node_type}: "
f"{list(new_found.keys())}"
)
self._write_pending(new_found)
return new_found
def _write_pending(self, new_fields: dict):
existing = {}
if os.path.exists(PENDING_FILE):
try:
existing = json.loads(open(PENDING_FILE,
encoding="utf-8").read())
except Exception:
existing = {}
existing.update(new_fields)
os.makedirs(os.path.dirname(PENDING_FILE), exist_ok=True)
with open(PENDING_FILE, "w", encoding="utf-8") as f:
json.dump(existing, f, indent=2, ensure_ascii=False)
logger.success(
f"[SchemaLearner] Pending additions written -> "
f"data/processed/pending_schema_additions.json"
)
def get_pending(self) -> dict:
if not os.path.exists(PENDING_FILE):
return {}
try:
return json.loads(open(PENDING_FILE, encoding="utf-8").read())
except Exception:
return {}
if __name__ == "__main__":
print("=" * 55)
print("BharatGraph - Schema Learner Test")
print("=" * 55)
learner = SchemaLearner()
sample = [
{"id": "p001", "name": "Test Politician",
"state": "Maharashtra", "party": "Test Party",
"new_field_tax_returns": "filed",
"foreign_assets_crore": 12.5,
"spouse_income_crore": 3.2},
]
new = learner.detect_new_fields(sample, "Politician")
print(f"\n New fields found: {len(new)}")
for k, v in new.items():
print(f" {k}: {v['sample_value']}")
print(f"\n Pending file: {PENDING_FILE}")
print("\nDone!")
|