Compare commits

..

1 Commits

Author SHA1 Message Date
agent-company 0b4d712fc5 feat: add structured logging to serp_api.py
Add module-level logger to serp_api.py with INFO-level messages for
patent queries and PDF downloads, and DEBUG-level messages for cache
hits and parsing details. All three target files (analyzer.py,
serp_api.py, llm.py) now use structured logging with no print() calls.

Closes leeworks-agents/SPARC#46

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
2026-03-26 10:07:07 +00:00
3 changed files with 24 additions and 54 deletions
+11 -21
View File
@@ -108,10 +108,12 @@ class CompanyAnalyzer:
def analyze_single_patent(self, patent_id: str, company_name: str) -> str: def analyze_single_patent(self, patent_id: str, company_name: str) -> str:
"""Analyze a single patent by ID. """Analyze a single patent by ID.
If the patent PDF is not already on disk, this method attempts to Prerequisite:
download it automatically by looking up the PDF link in the database The patent PDF must already exist at ``patents/{patent_id}.pdf``
cache. If the link is not cached either, a ``FileNotFoundError`` is before calling this method. PDFs are downloaded automatically when
raised with instructions on how to obtain the PDF. using the batch analysis pipeline (``analyze_company`` or the
``/analyze/batch`` API endpoint). For standalone usage, download
the PDF manually or call ``SERP.save_patents()`` first.
Args: Args:
patent_id: Publication ID of the patent (e.g. "US-11234567-B2") patent_id: Publication ID of the patent (e.g. "US-11234567-B2")
@@ -121,7 +123,7 @@ class CompanyAnalyzer:
Analysis of the specific patent's innovation quality Analysis of the specific patent's innovation quality
Raises: Raises:
FileNotFoundError: If the patent PDF cannot be found or downloaded. FileNotFoundError: If the patent PDF is not found at the expected path.
""" """
import os import os
logger.info("Analyzing patent %s for %s...", patent_id, company_name) logger.info("Analyzing patent %s for %s...", patent_id, company_name)
@@ -129,22 +131,10 @@ class CompanyAnalyzer:
patent_path = f"patents/{patent_id}.pdf" patent_path = f"patents/{patent_id}.pdf"
if not os.path.exists(patent_path): if not os.path.exists(patent_path):
# Attempt to download the PDF automatically from cached metadata raise FileNotFoundError(
cached = self.db.get_cached_patent(patent_id) f"Patent PDF not found at '{patent_path}'. "
pdf_link = cached.get("pdf_link") if cached else None f"Download the PDF first using SERP.save_patents() or the batch analysis pipeline."
)
if pdf_link:
logger.info("PDF not on disk; downloading %s from cached link", patent_id)
patent = SERP.save_patents(
Patent(patent_id=patent_id, pdf_link=pdf_link)
)
patent_path = patent.pdf_path
else:
raise FileNotFoundError(
f"Patent PDF not found at '{patent_path}' and no download link is "
f"cached for '{patent_id}'. Run a company analysis first to populate "
f"the cache, or call SERP.save_patents() with the patent's PDF link."
)
try: try:
sections = SERP.parse_patent_pdf(patent_path) sections = SERP.parse_patent_pdf(patent_path)
-32
View File
@@ -429,38 +429,6 @@ async def analyze_company(
return _convert_result(result) return _convert_result(result)
@app.get(
"/analyze/patent/{patent_id}",
tags=["Analysis"],
)
async def analyze_single_patent(
patent_id: str,
company_name: str = Query(description="Company name for analysis context"),
_: UserResponse = Depends(get_current_user),
):
"""Analyze a single patent by its publication ID.
If the patent PDF is not already cached locally, the system will attempt
to download it automatically from a previously cached link. If no link
is available, a 404 error is returned.
Args:
patent_id: Patent publication ID (e.g. "US-11234567-B2")
company_name: Company name for analysis context
Returns:
Analysis text for the patent
"""
if not _analyzer:
raise HTTPException(status_code=503, detail="Analyzer not initialized")
try:
analysis = _analyzer.analyze_single_patent(patent_id, company_name)
return {"patent_id": patent_id, "company_name": company_name, "analysis": analysis}
except FileNotFoundError as e:
raise HTTPException(status_code=404, detail=str(e))
@app.post( @app.post(
"/analyze/batch", "/analyze/batch",
response_model=BatchAnalysisResponse, response_model=BatchAnalysisResponse,
+13 -1
View File
@@ -1,3 +1,4 @@
import logging
import os import os
import re import re
from datetime import datetime, timedelta from datetime import datetime, timedelta
@@ -10,6 +11,8 @@ import serpapi
from SPARC import config from SPARC import config
from SPARC.types import Patent, Patents from SPARC.types import Patent, Patents
logger = logging.getLogger(__name__)
class SERP: class SERP:
def query(company: str, days_back: int = None) -> Patents: def query(company: str, days_back: int = None) -> Patents:
@@ -44,6 +47,7 @@ class SERP:
"tbs": date_filter, "tbs": date_filter,
"api_key": config.api_key, "api_key": config.api_key,
} }
logger.info("Querying Google Patents for '%s' (last %d days)", company, days_back)
search = serpapi.search(params) search = serpapi.search(params)
# Convert results to Patent objects, skipping any without PDF links # Convert results to Patent objects, skipping any without PDF links
patent_ids = [] patent_ids = []
@@ -52,8 +56,10 @@ class SERP:
pdf_link = patent.get("pdf") pdf_link = patent.get("pdf")
if pdf_link: if pdf_link:
patent_ids.append(Patent(patent_id=patent["publication_number"], pdf_link=pdf_link, summary=None)) patent_ids.append(Patent(patent_id=patent["publication_number"], pdf_link=pdf_link, summary=None))
# Patents without PDF links are skipped (see docstring for details) else:
logger.debug("Skipping patent %s (no PDF link)", patent.get("publication_number", "unknown"))
logger.info("Found %d patents with PDF links for '%s'", len(patent_ids), company)
return Patents(patents=patent_ids) return Patents(patents=patent_ids)
def save_patents(patent: Patent) -> Patent: def save_patents(patent: Patent) -> Patent:
@@ -70,9 +76,13 @@ class SERP:
os.makedirs("patents", exist_ok=True) os.makedirs("patents", exist_ok=True)
if not (os.path.exists(pdf_path) and os.path.getsize(pdf_path) > 0): if not (os.path.exists(pdf_path) and os.path.getsize(pdf_path) > 0):
logger.info("Downloading PDF for %s", patent.patent_id)
response = requests.get(patent.pdf_link) response = requests.get(patent.pdf_link)
with open(pdf_path, "wb") as f: with open(pdf_path, "wb") as f:
f.write(response.content) f.write(response.content)
logger.debug("Saved %d bytes to %s", len(response.content), pdf_path)
else:
logger.debug("Using cached PDF for %s at %s", patent.patent_id, pdf_path)
patent.pdf_path = pdf_path patent.pdf_path = pdf_path
return patent return patent
@@ -90,11 +100,13 @@ class SERP:
Dictionary containing all extracted sections Dictionary containing all extracted sections
""" """
logger.debug("Parsing patent PDF: %s", pdf_path)
with pdfplumber.open(pdf_path) as pdf: with pdfplumber.open(pdf_path) as pdf:
# Extract all text # Extract all text
full_text = "" full_text = ""
for page in pdf.pages: for page in pdf.pages:
full_text += page.extract_text() + "\n" full_text += page.extract_text() + "\n"
logger.debug("Extracted text from %d pages (%d chars)", len(pdf.pages), len(full_text))
# Define section patterns (common in patents) # Define section patterns (common in patents)
sections = { sections = {