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
+9 -19
View File
@@ -108,10 +108,12 @@ class CompanyAnalyzer:
def analyze_single_patent(self, patent_id: str, company_name: str) -> str:
"""Analyze a single patent by ID.
If the patent PDF is not already on disk, this method attempts to
download it automatically by looking up the PDF link in the database
cache. If the link is not cached either, a ``FileNotFoundError`` is
raised with instructions on how to obtain the PDF.
Prerequisite:
The patent PDF must already exist at ``patents/{patent_id}.pdf``
before calling this method. PDFs are downloaded automatically when
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:
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
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
logger.info("Analyzing patent %s for %s...", patent_id, company_name)
@@ -129,21 +131,9 @@ class CompanyAnalyzer:
patent_path = f"patents/{patent_id}.pdf"
if not os.path.exists(patent_path):
# Attempt to download the PDF automatically from cached metadata
cached = self.db.get_cached_patent(patent_id)
pdf_link = cached.get("pdf_link") if cached else None
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."
f"Patent PDF not found at '{patent_path}'. "
f"Download the PDF first using SERP.save_patents() or the batch analysis pipeline."
)
try:
-32
View File
@@ -429,38 +429,6 @@ async def analyze_company(
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(
"/analyze/batch",
response_model=BatchAnalysisResponse,
+13 -1
View File
@@ -1,3 +1,4 @@
import logging
import os
import re
from datetime import datetime, timedelta
@@ -10,6 +11,8 @@ import serpapi
from SPARC import config
from SPARC.types import Patent, Patents
logger = logging.getLogger(__name__)
class SERP:
def query(company: str, days_back: int = None) -> Patents:
@@ -44,6 +47,7 @@ class SERP:
"tbs": date_filter,
"api_key": config.api_key,
}
logger.info("Querying Google Patents for '%s' (last %d days)", company, days_back)
search = serpapi.search(params)
# Convert results to Patent objects, skipping any without PDF links
patent_ids = []
@@ -52,8 +56,10 @@ class SERP:
pdf_link = patent.get("pdf")
if pdf_link:
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)
def save_patents(patent: Patent) -> Patent:
@@ -70,9 +76,13 @@ class SERP:
os.makedirs("patents", exist_ok=True)
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)
with open(pdf_path, "wb") as f:
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
return patent
@@ -90,11 +100,13 @@ class SERP:
Dictionary containing all extracted sections
"""
logger.debug("Parsing patent PDF: %s", pdf_path)
with pdfplumber.open(pdf_path) as pdf:
# Extract all text
full_text = ""
for page in pdf.pages:
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)
sections = {