"""pdb_client — client Python minimal pour la PDB API (SEC Project Intelligence Database, UQO). Aucune dépendance obligatoire (urllib). pandas est optionnel pour `to_dataframe`. from pdb_client import PDB pdb = PDB("VOTRE_CLE") # ou variable d'environnement PDB_API_KEY pdb.stats()["overview"] for p in pdb.iter_projects(type="data_center", min_amount=5e8): print(p["ticker"], p["project_name"], p["total_amount_usd"]) df = pdb.to_dataframe(pdb.iter_projects(sector="Utilities")) pdb.sql("select project_type, count(*) n from projects group by 1 order by n desc") """ from __future__ import annotations import csv import io import json import os import time import urllib.error import urllib.parse import urllib.request from typing import Any, Iterator __version__ = "1.0.0" DEFAULT_BASE = "https://www.pdb-api.co/v1" class PDBError(RuntimeError): def __init__(self, status: int, detail: str, url: str): super().__init__(f"HTTP {status} — {detail} ({url})") self.status, self.detail, self.url = status, detail, url class PDB: def __init__(self, api_key: str | None = None, base_url: str = DEFAULT_BASE, timeout: float = 60.0, retries: int = 3): self.api_key = api_key or os.environ.get("PDB_API_KEY", "") if not self.api_key: raise ValueError("Clé d'API manquante : PDB(api_key=...) ou variable PDB_API_KEY.") self.base_url = base_url.rstrip("/") self.timeout = timeout self.retries = retries # ------------------------------------------------------------ transport def request(self, method: str, path: str, params: dict | None = None, body: dict | None = None, raw: bool = False) -> Any: q = {k: (str(v).lower() if isinstance(v, bool) else v) for k, v in (params or {}).items() if v is not None and v != ""} url = f"{self.base_url}/{path.lstrip('/')}" + (f"?{urllib.parse.urlencode(q)}" if q else "") data = json.dumps(body).encode() if body is not None else None headers = {"X-API-Key": self.api_key, "Accept": "application/json", "User-Agent": f"pdb_client/{__version__}"} if data is not None: headers["Content-Type"] = "application/json" for attempt in range(self.retries + 1): req = urllib.request.Request(url, data=data, method=method, headers=headers) try: with urllib.request.urlopen(req, timeout=self.timeout) as r: payload = r.read() return payload.decode() if raw else json.loads(payload) except urllib.error.HTTPError as e: detail = e.read().decode(errors="replace") try: detail = json.loads(detail).get("detail", detail) except Exception: pass if e.code in (429, 502, 503, 504) and attempt < self.retries: time.sleep(1.5 * (attempt + 1)) continue raise PDBError(e.code, detail, url) from None def get(self, path: str, **params) -> Any: return self.request("GET", path, params) # ------------------------------------------------------------ découverte def health(self) -> dict: return self.get("health") def stats(self) -> dict: return self.get("stats") def taxonomy(self) -> list[dict]: return self.get("taxonomy") def sectors(self) -> list[dict]: return self.get("sectors") def technologies(self, q: str | None = None, limit: int = 50) -> list[dict]: return self.get("technologies", q=q, limit=limit) def locations(self, q: str | None = None, limit: int = 50) -> list[dict]: return self.get("locations", q=q, limit=limit) def partners(self, q: str | None = None, limit: int = 50) -> list[dict]: return self.get("partners", q=q, limit=limit) def schema(self) -> list[dict]: return self.get("schema") # ------------------------------------------------------------ projets def projects(self, **filters) -> dict: """Une page : {total, limit, offset, items}. Filtres : q, type, sector, status, ticker, cik, location, tech, partner, min_amount, max_amount, year_from, year_to, min_confidence, has_amount, sort, order, limit, offset.""" return self.get("projects", **filters) def iter_projects(self, page_size: int = 500, max_items: int | None = None, **filters) -> Iterator[dict]: """Itère sur tous les projets correspondant aux filtres (pagination automatique).""" yield from self._paginate("projects", page_size, max_items, **filters) def project(self, project_id: str) -> dict: """Fiche complète : {project, timeline, mentions, graph, similar}.""" return self.get(f"projects/{project_id}") def similar(self, project_id: str, k: int = 10) -> list[dict]: return self.get(f"projects/{project_id}/similar", k=k) def projects_csv(self, **filters) -> str: return self.request("GET", "projects/export.csv", filters, raw=True) # ------------------------------------------------------------ mentions / sections def mentions(self, **filters) -> dict: return self.get("mentions", **filters) def iter_mentions(self, page_size: int = 500, max_items: int | None = None, **filters) -> Iterator[dict]: yield from self._paginate("mentions", page_size, max_items, **filters) def mention(self, mention_id: str) -> dict: return self.get(f"mentions/{mention_id}") def section(self, section_id: str, highlight: str | None = None) -> dict: return self.get(f"sections/{section_id}", highlight=highlight) # ------------------------------------------------------------ entreprises def companies(self, **filters) -> dict: return self.get("companies", **filters) def iter_companies(self, page_size: int = 500, **filters) -> Iterator[dict]: yield from self._paginate("companies", page_size, None, **filters) def company(self, ticker_or_cik: str) -> dict: return self.get(f"companies/{ticker_or_cik}") # ------------------------------------------------------------ graphe / recherche / SQL def graph_search(self, q: str, type: str | None = None, limit: int = 30) -> list[dict]: return self.get("graph/search", q=q, type=type, limit=limit) def graph_node(self, node_id: str, limit: int = 200) -> dict: return self.get(f"graph/node/{urllib.parse.quote(node_id, safe=':')}", limit=limit) def semantic(self, q: str, k: int = 20, type: str | None = None, sector: str | None = None) -> list[dict]: return self.get("search/semantic", q=q, k=k, type=type, sector=sector)["items"] def sql(self, query: str, limit: int = 500) -> dict: """Requête de lecture DuckDB. Retour : {columns, rows, truncated, elapsed_ms}.""" return self.request("POST", "sql", body={"sql": query, "limit": limit}) def sql_records(self, query: str, limit: int = 500) -> list[dict]: r = self.sql(query, limit) return [dict(zip(r["columns"], row)) for row in r["rows"]] # ------------------------------------------------------------ utilitaires def _paginate(self, path: str, page_size: int, max_items: int | None, **filters) -> Iterator[dict]: offset, n = int(filters.pop("offset", 0) or 0), 0 page_size = max(1, min(page_size, 500)) while True: page = self.get(path, limit=page_size, offset=offset, **filters) items = page.get("items", []) for it in items: yield it n += 1 if max_items and n >= max_items: return offset += len(items) if not items or offset >= page.get("total", 0): return @staticmethod def to_dataframe(records): """Convertit une liste/itérateur de dicts en DataFrame pandas (listes jointes par ' | ').""" import pandas as pd # optionnel rows = [] for r in records: rows.append({k: (" | ".join(map(str, v)) if isinstance(v, list) else v) for k, v in r.items()}) return pd.DataFrame(rows) @staticmethod def to_csv(records, path: str) -> int: rows = list(records) if not rows: return 0 with open(path, "w", newline="", encoding="utf-8") as f: w = csv.DictWriter(f, fieldnames=list(rows[0].keys())) w.writeheader() for r in rows: w.writerow({k: (" | ".join(map(str, v)) if isinstance(v, list) else v) for k, v in r.items()}) return len(rows) if __name__ == "__main__": # petit test : python pdb_client.py VOTRE_CLE import sys c = PDB(sys.argv[1] if len(sys.argv) > 1 else None) print(json.dumps(c.health(), indent=2)) ov = c.stats()["overview"] print(f"{ov['projects']:,} projets, {ov['mentions']:,} mentions, {ov['companies']} entreprises") for p in c.iter_projects(type="data_center", min_amount=5e8, sort="amount", max_items=5): print(f" {p['ticker']:6s} {p['project_name'][:50]:50s} {p['total_amount_usd']/1e9:6.1f} G$")