root/MarketingTool
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

promote source governance fields to first-class columns
CI / compose (push) Failing after 5m35s
Details

Browse Source
This commit is contained in:
Marco0300
2026-09-04 21:02:53 +02:00
parent 6aefbdc1f3
commit 4ec9dc608b
3 changed files with 39 additions and 6 deletions
Download Patch File Download Diff File Expand all files Collapse all files
apps/api/app/main.py
+20 -6
View File
@@ -89,7 +89,7 @@ def _initialize_database(db_path: str) -> sqlite3.Connection:
"pipeline_entries": (("notes", "TEXT NOT NULL DEFAULT ''"), ("next_action", "TEXT NOT NULL DEFAULT ''"), ("follow_up_at", "TEXT"), ("actor_user_id", "INTEGER"), ("idempotency_key", "TEXT"), ("version", "INTEGER NOT NULL DEFAULT 1")),
"interactions": (("outcome", "TEXT NOT NULL DEFAULT 'other'"), ("notes", "TEXT NOT NULL DEFAULT ''"), ("next_action", "TEXT NOT NULL DEFAULT ''"), ("follow_up_at", "TEXT"), ("actor_user_id", "INTEGER"), ("idempotency_key", "TEXT")),
"suppressions": (("active", "INTEGER NOT NULL DEFAULT 1"), ("updated_at", "TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP"), ("actor_user_id", "INTEGER")),
"sources": (("source_code", "TEXT NOT NULL DEFAULT ''"), ("display_name", "TEXT NOT NULL DEFAULT ''"), ("approved", "INTEGER NOT NULL DEFAULT 0"), ("policy_json", "TEXT NOT NULL DEFAULT '{}'"), ("quota_json", "TEXT NOT NULL DEFAULT '{}'")),
"sources": (("source_code", "TEXT NOT NULL DEFAULT ''"), ("display_name", "TEXT NOT NULL DEFAULT ''"), ("approved", "INTEGER NOT NULL DEFAULT 0"), ("policy_json", "TEXT NOT NULL DEFAULT '{}'"), ("quota_json", "TEXT NOT NULL DEFAULT '{}'"), ("owner", "TEXT NOT NULL DEFAULT ''"), ("terms_url", "TEXT NOT NULL DEFAULT ''"), ("terms_status", "TEXT NOT NULL DEFAULT 'unreviewed'"), ("rate_limit", "TEXT NOT NULL DEFAULT ''"), ("daily_quota", "INTEGER"), ("credentials_configured", "INTEGER NOT NULL DEFAULT 0")),
"discovery_queries": (("selected_adapters_json", "TEXT NOT NULL DEFAULT '[]'"), ("location", "TEXT NOT NULL DEFAULT ''"), ("category", "TEXT NOT NULL DEFAULT ''"), ("max_records", "INTEGER NOT NULL DEFAULT 100"), ("daily_limit", "INTEGER NOT NULL DEFAULT 1000"), ("schedule", "TEXT NOT NULL DEFAULT ''"), ("dry_run", "INTEGER NOT NULL DEFAULT 0"), ("lifecycle", "TEXT NOT NULL DEFAULT 'draft'")),
"discovery_runs": (("selected_adapters_json", "TEXT NOT NULL DEFAULT '[]'"), ("location", "TEXT NOT NULL DEFAULT ''"), ("category", "TEXT NOT NULL DEFAULT ''"), ("max_records", "INTEGER NOT NULL DEFAULT 100"), ("daily_limit", "INTEGER NOT NULL DEFAULT 1000"), ("schedule", "TEXT NOT NULL DEFAULT ''"), ("dry_run", "INTEGER NOT NULL DEFAULT 0"), ("lifecycle", "TEXT NOT NULL DEFAULT 'draft'"), ("paused_at", "TEXT")),
"source_records": (("discovery_run_id", "INTEGER"), ("normalized_key", "TEXT NOT NULL DEFAULT ''"), ("provenance_json", "TEXT NOT NULL DEFAULT '{}'"), ("response_metadata_json", "TEXT NOT NULL DEFAULT '{}'")),
@@ -107,6 +107,7 @@ def _initialize_database(db_path: str) -> sqlite3.Connection:
CREATE TABLE sources_rebuilt (
id INTEGER PRIMARY KEY AUTOINCREMENT, organization_id TEXT NOT NULL REFERENCES organizations(id),
name TEXT NOT NULL, kind TEXT NOT NULL CHECK(kind IN ('csv','manual','google_places','google_browser_search','bing_local','approved_directory','public_website','permitted_social','ct_logs','dns','rdap')), source_code TEXT NOT NULL DEFAULT '', display_name TEXT NOT NULL DEFAULT '', enabled INTEGER NOT NULL DEFAULT 0, approved INTEGER NOT NULL DEFAULT 0,
owner TEXT NOT NULL DEFAULT '', terms_url TEXT NOT NULL DEFAULT '', terms_status TEXT NOT NULL DEFAULT 'unreviewed', rate_limit TEXT NOT NULL DEFAULT '', daily_quota INTEGER, credentials_configured INTEGER NOT NULL DEFAULT 0,
config_json TEXT NOT NULL DEFAULT '{}', policy_json TEXT NOT NULL DEFAULT '{}', quota_json TEXT NOT NULL DEFAULT '{}', health_status TEXT NOT NULL DEFAULT 'unknown',
consecutive_failures INTEGER NOT NULL DEFAULT 0, circuit_open INTEGER NOT NULL DEFAULT 0,
last_success_at TEXT, last_failure_at TEXT, last_error TEXT, created_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP,
@@ -1449,7 +1450,7 @@ class ApiHandler(BaseHTTPRequestHandler):
else:seen.add(key);accepted.append(b)
return self.send_json(200,{"accepted":len(accepted),"duplicates":len(rows)-len(normalized)+len(normalized)-len(accepted)-suppressed,"suppressed":suppressed,"rows":accepted})
def list_sources(self,db,org):
cols='id,organization_id,name,kind,source_code,display_name,enabled,approved,config_json,policy_json,quota_json,health_status,consecutive_failures,circuit_open,last_success_at,last_failure_at,last_error,created_at,updated_at'
cols='*'
items=[]
adapter_meta={a["source_code"]:a for a in available_adapters()}
for raw in db.execute(f"SELECT {cols} FROM sources WHERE organization_id=? ORDER BY id",(org,)):
@@ -1468,9 +1469,12 @@ class ApiHandler(BaseHTTPRequestHandler):
# back through setup.
"configured": bool(meta.get("available", False) and (config.get("csv") or config.get("rows") is not None or raw["approved"] or config.get("approved") is True)),
"credential_status": ("Configured" if credential_exists else "Required / not configured") if meta.get("requires_credentials") else "Not required",
"terms_status": "Provided" if policy.get("terms_url") or config.get("terms_url") else "Not reviewed",
"owner": policy.get("owner") or config.get("owner") or "Not assigned",
"rate_limit": policy.get("rate_limit") or config.get("rate_limit") or "Not set"})
"credentials_configured": credential_exists,
"terms_url": raw["terms_url"] or policy.get("terms_url") or config.get("terms_url") or "",
"terms_status": raw["terms_status"] or policy.get("terms_status") or config.get("terms_status") or "unreviewed",
"owner": raw["owner"] or policy.get("owner") or config.get("owner") or "Not assigned",
"rate_limit": raw["rate_limit"] or policy.get("rate_limit") or config.get("rate_limit") or "Not set",
"daily_quota": raw["daily_quota"] if raw["daily_quota"] is not None else quota.get("daily_quota", config.get("daily_quota"))})
item.pop("config_json", None)
items.append(item)
return self.send_json(200,{"organization_id":org,"items":items})
@@ -1507,12 +1511,22 @@ class ApiHandler(BaseHTTPRequestHandler):
if requested == 'google_browser_search' and bool(payload.get('enabled', False)) and os.environ.get('GOOGLE_BROWSER_SEARCH_ENABLED', '').strip().lower() != 'true': return self.send_json(409,{"error":"source_feature_disabled","detail":"GOOGLE_BROWSER_SEARCH_ENABLED=true is required"})
if contains_secret(config):return self.send_json(400,{"error":"secret_not_permitted"})
if any(field in config for field in DISCOVERY_CRITERIA_FIELDS):return self.send_json(400,{"error":"source_configuration_contains_criteria"})
policy = payload.get('policy', {}) if isinstance(payload.get('policy', {}), dict) else {}
quota = payload.get('quota', {}) if isinstance(payload.get('quota', {}), dict) else {}
owner = str(policy.get('owner', config.get('owner', ''))).strip()
terms_url = str(policy.get('terms_url', config.get('terms_url', ''))).strip()
terms_status = str(policy.get('terms_status', config.get('terms_status', 'unreviewed'))).strip().lower() or 'unreviewed'
rate_limit = str(policy.get('rate_limit', config.get('rate_limit', ''))).strip()
daily_quota = quota.get('daily_quota', config.get('daily_quota'))
try: daily_quota = int(daily_quota) if daily_quota not in (None, '') else None
except (TypeError, ValueError): return self.send_json(400, {'error':'invalid_source_quota'})
if daily_quota is not None and not 1 <= daily_quota <= 100000: return self.send_json(400, {'error':'invalid_source_quota'})
try:
validation=adapter_for(source_code).validate_config(config)
# Registration may precede the actual local payload or optional provider
# approval. Discovery/ingest still validates the effective configuration.
if config and not validation.valid:return self.send_json(400,{"error":"invalid_source_config","details":validation.errors})
cur=db.execute("INSERT INTO sources(organization_id,name,kind,source_code,display_name,enabled,approved,config_json,policy_json,quota_json) VALUES(?,?,?,?,?,?,?,?,?,?)",(user['organization_id'],name,kind,source_code,str(payload.get('display_name') or adapter_for(source_code).display_name),int(bool(payload.get('enabled',False))),int(bool(payload.get('approved',config.get('approved',False)))),json.dumps(config,sort_keys=True),json.dumps(payload.get('policy',{}),sort_keys=True),json.dumps(payload.get('quota',{}),sort_keys=True)))
cur=db.execute("INSERT INTO sources(organization_id,name,kind,source_code,display_name,enabled,approved,owner,terms_url,terms_status,rate_limit,daily_quota,credentials_configured,config_json,policy_json,quota_json) VALUES(?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)",(user['organization_id'],name,kind,source_code,str(payload.get('display_name') or adapter_for(source_code).display_name),int(bool(payload.get('enabled',False))),int(bool(payload.get('approved',config.get('approved',False)))),owner,terms_url,terms_status,rate_limit,daily_quota,0,json.dumps(config,sort_keys=True),json.dumps(policy,sort_keys=True),json.dumps(quota,sort_keys=True)))
except sqlite3.IntegrityError:
existing=db.execute("SELECT id,organization_id,name,kind,source_code,display_name,enabled,approved,policy_json,quota_json,health_status,consecutive_failures,circuit_open,last_success_at,last_failure_at,last_error,created_at,updated_at FROM sources WHERE organization_id=? AND name=?",(user['organization_id'],name)).fetchone()
if existing:
apps/api/schema.sql
+1
View File
@@ -137,6 +137,7 @@ CREATE INDEX IF NOT EXISTS idx_job_events_job_sequence ON job_events(job_id,sequ
CREATE TABLE IF NOT EXISTS sources (
id INTEGER PRIMARY KEY AUTOINCREMENT, organization_id TEXT NOT NULL REFERENCES organizations(id),
name TEXT NOT NULL, kind TEXT NOT NULL CHECK(kind IN ('csv','manual','google_places','google_browser_search','bing_local','approved_directory','public_website','permitted_social','ct_logs','dns','rdap')), source_code TEXT NOT NULL DEFAULT '', display_name TEXT NOT NULL DEFAULT '', enabled INTEGER NOT NULL DEFAULT 0, approved INTEGER NOT NULL DEFAULT 0,
owner TEXT NOT NULL DEFAULT '', terms_url TEXT NOT NULL DEFAULT '', terms_status TEXT NOT NULL DEFAULT 'unreviewed', rate_limit TEXT NOT NULL DEFAULT '', daily_quota INTEGER, credentials_configured INTEGER NOT NULL DEFAULT 0,
config_json TEXT NOT NULL DEFAULT '{}', policy_json TEXT NOT NULL DEFAULT '{}', quota_json TEXT NOT NULL DEFAULT '{}', health_status TEXT NOT NULL DEFAULT 'unknown',
consecutive_failures INTEGER NOT NULL DEFAULT 0, circuit_open INTEGER NOT NULL DEFAULT 0,
last_success_at TEXT, last_failure_at TEXT, last_error TEXT, created_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP,
apps/api/tests/test_sources_phase5.py
+18
View File
@@ -113,6 +113,24 @@ class SourceApiTests(unittest.TestCase):
second={**payload,'name':'OpenStreetMap / Overpass · plumbers · Durban'}
status, other=self.req('POST','/api/v1/sources',second); self.assertEqual(status,201); self.assertTrue(other['created']); self.assertNotEqual(other['id'],created['id'])
def test_source_registry_exposes_first_class_governance_fields(self):
payload = {
'name': 'Governed manual', 'kind': 'manual', 'config': {'rows': []},
'policy': {'owner': 'Research Ops', 'terms_url': 'https://example.test/terms', 'terms_status': 'approved', 'rate_limit': '60/hour'},
'quota': {'daily_quota': 200},
}
status, created = self.req('POST', '/api/v1/sources', payload)
self.assertEqual(status, 201)
status, items = self.req('GET', '/api/v1/sources')
self.assertEqual(status, 200)
source = next(item for item in items['items'] if item['id'] == created['id'])
self.assertEqual(source['owner'], 'Research Ops')
self.assertEqual(source['terms_url'], 'https://example.test/terms')
self.assertEqual(source['terms_status'], 'approved')
self.assertEqual(source['rate_limit'], '60/hour')
self.assertEqual(source['daily_quota'], 200)
self.assertFalse(source['credentials_configured'])
def test_source_configuration_rejects_discovery_criteria(self):
forbidden = {'query': 'plumbers', 'category': 'trades', 'city': 'Cape Town', 'location': 'Western Cape'}
for field, value in forbidden.items():