233 lines
8.5 KiB
Python
233 lines
8.5 KiB
Python
"""
|
|
Political Article Filter — scans newspaper PDF pages using Claude Vision
|
|
to identify politically significant articles that opposition can use.
|
|
|
|
Run after PDF is uploaded and pages are rendered to images.
|
|
|
|
Usage:
|
|
python political_filter.py # processes latest job
|
|
python political_filter.py 20260526_012054 # specific job
|
|
"""
|
|
import os
|
|
import sys
|
|
import json
|
|
import base64
|
|
import anthropic
|
|
from pathlib import Path
|
|
|
|
OUTPUT_DIR = Path(__file__).parent / "output"
|
|
|
|
|
|
def encode_image(img_path):
|
|
with open(img_path, "rb") as f:
|
|
return base64.standard_b64encode(f.read()).decode("utf-8")
|
|
|
|
|
|
def scan_pages_for_political_articles(client, page_images, model="claude-sonnet-4-6"):
|
|
"""Send all page images to Claude and get politically significant articles."""
|
|
|
|
content = []
|
|
for i, img_path in enumerate(page_images, 1):
|
|
content.append({
|
|
"type": "image",
|
|
"source": {
|
|
"type": "base64",
|
|
"media_type": "image/png",
|
|
"data": encode_image(img_path),
|
|
},
|
|
})
|
|
content.append({
|
|
"type": "text",
|
|
"text": f"[This is page {i} of the newspaper]"
|
|
})
|
|
|
|
content.append({
|
|
"type": "text",
|
|
"text": """You are a political analyst working for an opposition party in Telangana, India.
|
|
|
|
Scan ALL pages of this Telugu newspaper carefully. Identify EVERY article that is politically significant — specifically issues that an opposition party can use for political mileage.
|
|
|
|
INCLUDE articles about:
|
|
- MLA/MLC/MP statements, press conferences, and speeches
|
|
- Political party activities, padayatras, and protests
|
|
- Government scheme announcements tied to politicians
|
|
- Collector administrative directives or warnings
|
|
- Government failures, delays, broken promises
|
|
- Corruption, scams, misuse of funds by ruling party leaders or officials
|
|
- Price hikes, inflation affecting common people
|
|
- Farmer distress, crop losses, loan issues
|
|
- Infrastructure failures (roads, water, electricity, hospitals)
|
|
- Law and order failures, crime, police issues
|
|
- Unemployment, job losses
|
|
- Public grievances against government departments
|
|
- Controversial government policies or decisions
|
|
- Ruling party internal conflicts or embarrassments
|
|
- Any issue that can be raised in assembly, press conference, or public rally
|
|
|
|
IGNORE articles about:
|
|
- Sports, entertainment, cinema
|
|
- Horoscopes, puzzles, classifieds
|
|
- International news with no local political angle
|
|
- Advertisements
|
|
|
|
For each politically significant article, provide:
|
|
1. page_number: which page it's on
|
|
2. headline: the Telugu headline (copy exactly from image)
|
|
3. headline_english: English translation of headline
|
|
4. location: describe where on the page (top-left, center, bottom-right, etc.)
|
|
5. political_significance: why this matters for opposition (2-3 sentences)
|
|
6. attack_angle: how opposition can use this (1-2 sentences)
|
|
7. priority: "high", "medium", or "low"
|
|
8. category: one of ["corruption", "governance_failure", "public_grievance", "economy", "law_order", "health", "education", "infrastructure", "ruling_party_crisis", "policy_controversy", "other"]
|
|
|
|
Return ONLY valid JSON array. No markdown, no explanation. Example:
|
|
[
|
|
{
|
|
"page_number": 1,
|
|
"headline": "Telugu headline here",
|
|
"headline_english": "English translation",
|
|
"location": "top-right, large article with photo",
|
|
"political_significance": "Why this matters...",
|
|
"attack_angle": "How to use this...",
|
|
"priority": "high",
|
|
"category": "corruption"
|
|
}
|
|
]
|
|
|
|
Be thorough. Don't miss any politically useful article. A good opposition analyst finds ammunition in everything."""
|
|
})
|
|
|
|
response = client.messages.create(
|
|
model=model,
|
|
max_tokens=8192,
|
|
messages=[{"role": "user", "content": content}],
|
|
)
|
|
|
|
response_text = response.content[0].text.strip()
|
|
# Clean up if wrapped in markdown
|
|
if response_text.startswith("```"):
|
|
response_text = response_text.split("\n", 1)[1]
|
|
if response_text.endswith("```"):
|
|
response_text = response_text[:-3]
|
|
|
|
return json.loads(response_text)
|
|
|
|
|
|
def process_job(job_id=None):
|
|
api_key = os.environ.get("ANTHROPIC_API_KEY")
|
|
if not api_key:
|
|
print("Error: Set ANTHROPIC_API_KEY environment variable")
|
|
sys.exit(1)
|
|
|
|
# Find job directory
|
|
if job_id:
|
|
job_dir = OUTPUT_DIR / job_id
|
|
else:
|
|
jobs = sorted(OUTPUT_DIR.iterdir(), reverse=True)
|
|
if not jobs:
|
|
print("No jobs found")
|
|
sys.exit(1)
|
|
job_dir = jobs[0]
|
|
job_id = job_dir.name
|
|
|
|
pages_dir = job_dir / "pages"
|
|
if not pages_dir.exists():
|
|
print(f"No pages found in {job_dir}")
|
|
sys.exit(1)
|
|
|
|
# Collect page images
|
|
page_images = sorted(pages_dir.glob("page_*.png"))
|
|
if not page_images:
|
|
print("No page images found")
|
|
sys.exit(1)
|
|
|
|
print(f"Job: {job_id}")
|
|
print(f"Pages: {len(page_images)}")
|
|
print("Scanning for politically significant articles...")
|
|
print()
|
|
|
|
client = anthropic.Anthropic(api_key=api_key)
|
|
articles = scan_pages_for_political_articles(client, page_images)
|
|
|
|
# Sort by priority
|
|
priority_order = {"high": 0, "medium": 1, "low": 2}
|
|
articles.sort(key=lambda a: priority_order.get(a.get("priority", "low"), 3))
|
|
|
|
# Save results
|
|
output_path = job_dir / "political_articles.json"
|
|
with open(output_path, "w", encoding="utf-8") as f:
|
|
json.dump(articles, f, indent=2, ensure_ascii=False)
|
|
|
|
# Print summary
|
|
print(f"Found {len(articles)} politically significant articles:")
|
|
print("=" * 70)
|
|
|
|
high = [a for a in articles if a.get("priority") == "high"]
|
|
medium = [a for a in articles if a.get("priority") == "medium"]
|
|
low = [a for a in articles if a.get("priority") == "low"]
|
|
|
|
print(f" HIGH priority: {len(high)}")
|
|
print(f" MEDIUM priority: {len(medium)}")
|
|
print(f" LOW priority: {len(low)}")
|
|
print()
|
|
|
|
for i, article in enumerate(articles, 1):
|
|
priority_tag = f"[{article.get('priority', '?').upper()}]"
|
|
category = article.get("category", "other")
|
|
print(f"{i}. {priority_tag} [{category}] (Page {article.get('page_number', '?')})")
|
|
print(f" {article.get('headline', 'N/A')}")
|
|
print(f" ({article.get('headline_english', 'N/A')})")
|
|
print(f" Political angle: {article.get('attack_angle', 'N/A')}")
|
|
print()
|
|
|
|
# Generate brief text file
|
|
brief_path = job_dir / "opposition_daily_brief.txt"
|
|
with open(brief_path, "w", encoding="utf-8") as f:
|
|
f.write("=" * 70 + "\n")
|
|
f.write(" OPPOSITION DAILY BRIEF\n")
|
|
f.write(f" Source: {job_id}\n")
|
|
f.write("=" * 70 + "\n\n")
|
|
|
|
f.write(f"Total politically significant articles: {len(articles)}\n")
|
|
f.write(f"HIGH priority: {len(high)} | MEDIUM: {len(medium)} | LOW: {len(low)}\n")
|
|
f.write("-" * 70 + "\n\n")
|
|
|
|
if high:
|
|
f.write(">>> HIGH PRIORITY — IMMEDIATE ACTION NEEDED <<<\n\n")
|
|
for i, a in enumerate(high, 1):
|
|
f.write(f"{i}. {a.get('headline', 'N/A')}\n")
|
|
f.write(f" ({a.get('headline_english', 'N/A')})\n")
|
|
f.write(f" Page: {a.get('page_number', '?')} | Category: {a.get('category', '?')}\n")
|
|
f.write(f" WHY IT MATTERS: {a.get('political_significance', 'N/A')}\n")
|
|
f.write(f" ATTACK ANGLE: {a.get('attack_angle', 'N/A')}\n\n")
|
|
|
|
if medium:
|
|
f.write("-" * 70 + "\n")
|
|
f.write(">>> MEDIUM PRIORITY <<<\n\n")
|
|
for i, a in enumerate(medium, 1):
|
|
f.write(f"{i}. {a.get('headline', 'N/A')}\n")
|
|
f.write(f" ({a.get('headline_english', 'N/A')})\n")
|
|
f.write(f" Page: {a.get('page_number', '?')} | Category: {a.get('category', '?')}\n")
|
|
f.write(f" WHY IT MATTERS: {a.get('political_significance', 'N/A')}\n")
|
|
f.write(f" ATTACK ANGLE: {a.get('attack_angle', 'N/A')}\n\n")
|
|
|
|
if low:
|
|
f.write("-" * 70 + "\n")
|
|
f.write(">>> LOW PRIORITY — TRACK FOR LATER <<<\n\n")
|
|
for i, a in enumerate(low, 1):
|
|
f.write(f"{i}. {a.get('headline', 'N/A')}\n")
|
|
f.write(f" ({a.get('headline_english', 'N/A')})\n")
|
|
f.write(f" ATTACK ANGLE: {a.get('attack_angle', 'N/A')}\n\n")
|
|
|
|
f.write("=" * 70 + "\n")
|
|
f.write("END OF BRIEF\n")
|
|
|
|
print("=" * 70)
|
|
print(f"Saved: {output_path}")
|
|
print(f"Saved: {brief_path}")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
job_id = sys.argv[1] if len(sys.argv) > 1 else None
|
|
process_job(job_id)
|