add Google Places source dashboard

apps/api/app/main.py

@@ -107,6 +107,15 @@ def row_json(row):
         if key in result: result[key] = bool(result[key])
     return result
 
+def source_config_with_credentials(db, source, config):
+    """Inject a credential only into the in-process adapter call; never persist/return it."""
+    if source["kind"] != "google_places": return config
+    row = db.execute("SELECT secret_ref FROM source_credentials WHERE source_id=? AND organization_id=? AND provider=? AND key_name=?", (source["id"], source["organization_id"], "google_places", "api_key")).fetchone()
+    if not row or not row["secret_ref"]: return config
+    try: config["_api_key"] = decrypt_provider_secret(row["secret_ref"])
+    except Exception: pass
+    return config
+
 class ApiHandler(BaseHTTPRequestHandler):
     server_version = "ProspectPlatform/0.1"
     def send_json(self, status, payload, extra_headers=None):
@@ -1095,6 +1104,9 @@ class ApiHandler(BaseHTTPRequestHandler):
             bits_outreach=path.split("/")
             if len(bits_outreach)==7 and bits_outreach[:4]==["","api","v1","outreach"] and bits_outreach[4]=="drafts" and bits_outreach[5].isdigit() and bits_outreach[6] in {"approve","send"}: return self.approve_outreach_draft(int(bits_outreach[5]),db,user) if bits_outreach[6]=="approve" else self.send_outreach_draft(int(bits_outreach[5]),db,user)
             if path=="/api/v1/sources":return self.create_source(payload,db,user)
+            if path.startswith("/api/v1/sources/") and path.endswith("/credentials"):
+                bits=path.split("/")
+                if len(bits)==6 and bits[4].isdigit(): return self.save_source_credential(int(bits[4]),payload,db,user)
             if path=="/api/v1/discovery":return self.create_scoped_discovery(payload,db,user)
             if path.startswith("/api/v1/discovery-runs/"):
                 bits=path.split("/")
@@ -1320,9 +1332,10 @@ class ApiHandler(BaseHTTPRequestHandler):
             except (TypeError,ValueError): config={}
             try: policy=json.loads(raw["policy_json"] or "{}")
             except (TypeError,ValueError): policy={}
+            credential_exists = bool(db.execute("SELECT 1 FROM source_credentials WHERE source_id=? AND organization_id=? AND provider=? AND key_name=?", (raw["id"], org, "google_places", "api_key")).fetchone())
             item.update({"available": bool(meta.get("available", False)), "optional": bool(meta.get("optional", False)),
                          "configured": bool(meta.get("available", False) and (config.get("csv") or config.get("rows") is not None or raw["approved"])),
-                         "credential_status": "Not required" if not meta.get("requires_credentials") else "Required / not configured",
+                         "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"})
@@ -1331,6 +1344,15 @@ class ApiHandler(BaseHTTPRequestHandler):
         return self.send_json(200,{"organization_id":org,"items":items})
     def list_queries(self,db,org):
         return self.send_json(200,{"organization_id":org,"items":[row_json(r) for r in db.execute("SELECT * FROM discovery_queries WHERE organization_id=? ORDER BY id",(org,))]})
+    def save_source_credential(self, sid, payload, db, user):
+        source=db.execute("SELECT * FROM sources WHERE id=? AND organization_id=?",(sid,user["organization_id"])).fetchone()
+        if not source: return self.send_json(404,{"error":"not_found"})
+        provider=str(payload.get("provider", "")).strip().lower(); key_name=str(payload.get("key_name", "api_key")).strip().lower(); value=payload.get("api_key")
+        if source["kind"] != "google_places" or provider != "google_places" or key_name != "api_key" or not isinstance(value,str) or not (8 <= len(value.strip()) <= 4096): return self.send_json(400,{"error":"invalid_source_credential"})
+        ciphertext=encrypt_provider_secret(value.strip())
+        db.execute("INSERT INTO source_credentials(source_id,organization_id,provider,key_name,secret_ref,metadata_json) VALUES(?,?,?,?,?,?) ON CONFLICT(source_id,provider,key_name) DO UPDATE SET secret_ref=excluded.secret_ref,metadata_json=excluded.metadata_json",(sid,user["organization_id"],provider,key_name,ciphertext,json.dumps({"configured_at":datetime.now(timezone.utc).replace(microsecond=0).isoformat()})))
+        self.audit(db,user,"source.credential.updated",str(sid)); db.commit()
+        return self.send_json(200,{"ok":True,"configured":True,"provider":provider,"key_name":key_name})
     def list_source_records(self,db,org,q):
         try:
             limit=int(q.get('page_size',[50])[0]); offset=max(0,int(q.get('offset',[0])[0]))
@@ -1382,6 +1404,8 @@ class ApiHandler(BaseHTTPRequestHandler):
             # before they are enabled.
             if not validation.valid and not (source['kind'] == 'manual' and not config):
                 return self.send_json(409,{"error":"source_not_configured","details":validation.errors})
+            if source['kind'] == 'google_places' and not db.execute("SELECT 1 FROM source_credentials WHERE source_id=? AND organization_id=? AND provider=? AND key_name=?", (sid,user['organization_id'],'google_places','api_key')).fetchone():
+                return self.send_json(409,{"error":"source_not_configured","details":["Google Places API key is required"]})
         db.execute("UPDATE sources SET enabled=?,updated_at=CURRENT_TIMESTAMP WHERE id=?",(value,sid));self.audit(db,user,'source.enabled' if value else 'source.disabled',str(sid));db.commit();return self.send_json(200,row_json(db.execute("SELECT * FROM sources WHERE id=?",(sid,)).fetchone()))
     def create_query(self,payload,db,user):
         sid=payload.get('source_id');name=str(payload.get('name','')).strip();query=payload.get('query',{})
@@ -1421,7 +1445,7 @@ class ApiHandler(BaseHTTPRequestHandler):
         if len(json.dumps(config).encode())>5*1024*1024:return self.send_json(400,{"error":"ingest_limits"})
         if isinstance(config.get('rows'),list) and (len(config['rows'])>1000 or any(not isinstance(r,dict) or len(r)>50 or any(len(str(v))>10000 for v in r.values()) for r in config['rows'])):return self.send_json(400,{"error":"ingest_limits"})
         if isinstance(config.get('csv'),str) and config['csv'].count('\n')>1001:return self.send_json(400,{"error":"ingest_limits"})
-        try:page=adapter_for(source['kind']).discover(config)
+        try:page=adapter_for(source['kind']).discover(source_config_with_credentials(db, source, config))
         except (ValueError,KeyError) as exc:return self.send_json(400,{"error":"invalid_ingest","detail":str(exc)})
         inserted=0
         for record in page.records[:1000]:
@@ -1559,7 +1583,7 @@ def _run_source_discovery(db, job, handler):
             used_today=db.execute("SELECT COUNT(*) FROM source_records WHERE organization_id=? AND source_id=? AND date(created_at)=date('now')",(org,source["id"])).fetchone()[0]
             if used_today >= daily_limit or per_run_limit <= 0:
                 handler.add_job_event(db,job["id"],org,"source.quota_exceeded",f"Quota reached for {source['kind']}",10,"SOURCE_QUOTA_EXCEEDED"); continue
-            page=adapter_for(source["kind"]).discover(config)
+            page=adapter_for(source["kind"]).discover(source_config_with_credentials(db, source, config))
         except Exception as exc:
             failures=int(source["consecutive_failures"])+1
             db.execute("UPDATE sources SET health_status='unhealthy',consecutive_failures=?,circuit_open=?,last_failure_at=CURRENT_TIMESTAMP,last_error=?,updated_at=CURRENT_TIMESTAMP WHERE id=?",(failures,int(circuit_is_open(failures)),str(exc)[:300],source["id"]))