engineering/pig.py
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Refactor language handling to FastAPI Depends

Browse source 
- Replace get_lang() calls with Depends(get_language) dependency
- Language resolution: ?lang=XX query > cookie > Accept-Language header
- Middleware sets cookie, frontend cleans URL via history.replaceState
- view_page_by_hash now redirects instead of direct call
This commit is contained in:
Russell Ballestrini 2026-01-05 22:36:55 -05:00
parent 4b247f0be9
commit 0974593873
4 changed files with 71 additions and 66 deletions
Download patch file Download diff file Expand all files Collapse all files

2
serp/__init__.py
View file

@ -15,7 +15,7 @@ from .i18n import (
SEARCH_BOX_HTML,
)
from .models import CrawlRequest
from .app import app, templates, DB_PATH, VAULT_PATH, CRAWL_DISABLED, IMPORT_MODE
from .app import app, templates, DB_PATH, VAULT_PATH, CRAWL_DISABLED, IMPORT_MODE, get_language
from .archive import (
ArchiveDB,
read_from_tarball,

72
serp/app.py
View file

@ -13,58 +13,72 @@ import logging
import os
from pathlib import Path
from fastapi import FastAPI, Request
from fastapi import FastAPI, Request, Cookie, Header
from fastapi.staticfiles import StaticFiles
from fastapi.templating import Jinja2Templates
from starlette.middleware.base import BaseHTTPMiddleware
from .i18n import TRANSLATIONS
logger = logging.getLogger(__name__)
# Create app
app = FastAPI(title="neopig", description="Media crawler + SERP + Screenshot service")
# Supported language codes
SUPPORTED_LANGS = set(TRANSLATIONS.keys())
class LanguageMiddleware(BaseHTTPMiddleware):
"""Middleware to handle ?lang=XX query param for language switching.
async def get_language(
request: Request,
lang: str = Cookie(None),
accept_language: str = Header(None)
) -> str:
"""Dependency: resolve language from ?lang=XX > cookie > Accept-Language.
Easter egg #0: Use ?lang=XX to force any page into a specific language.
Example: /?lang=ka forces Georgian, /?lang=ja forces Japanese.
When ?lang=XX is present:
1. Validates against supported languages
2. Sets the 'lang' cookie on response
3. Frontend JS cleans URL via history.replaceState (no redirect)
Cookie is set via middleware (LanguageCookieMiddleware).
"""
# Query param takes priority
lang_query = request.query_params.get("lang")
if lang_query and lang_query.lower() in SUPPORTED_LANGS:
return lang_query.lower()
# Supported language codes (must match TRANSLATIONS keys)
SUPPORTED_LANGS = {
"en", "zh", "zh-tw", "es", "hi", "ar", "pt", "ru", "ja", "fr", "de",
"ko", "it", "nl", "pl", "tr", "vi", "th", "id", "uk", "sv",
"bn", "ur", "sw", "mr", "te", "ka"
}
# Cookie next
if lang and lang in SUPPORTED_LANGS:
return lang
# Accept-Language header
if accept_language:
for part in accept_language.split(','):
code = part.split(';')[0].strip().split('-')[0].lower()
if code in SUPPORTED_LANGS:
return code
return "en"
from starlette.middleware.base import BaseHTTPMiddleware
class LanguageCookieMiddleware(BaseHTTPMiddleware):
"""Middleware to set lang cookie when ?lang=XX is used."""
async def dispatch(self, request: Request, call_next):
# Check for lang query param
lang_param = request.query_params.get("lang")
# Process request normally
response = await call_next(request)
# If valid lang param, set cookie (frontend cleans URL)
if lang_param and lang_param.lower() in self.SUPPORTED_LANGS:
# Set cookie if ?lang= query param is present and valid
lang_query = request.query_params.get("lang")
if lang_query and lang_query.lower() in SUPPORTED_LANGS:
response.set_cookie(
key="lang",
value=lang_param.lower(),
max_age=365 * 24 * 60 * 60, # 1 year
httponly=False, # Allow JS access for dropdown
value=lang_query.lower(),
max_age=365 * 24 * 60 * 60,
httponly=False,
samesite="lax"
)
return response
# Add language middleware
app.add_middleware(LanguageMiddleware)
app.add_middleware(LanguageCookieMiddleware)
# Paths relative to serp.py location (parent of this package)
BASE_PATH = Path(__file__).parent.parent

12
serp/i18n.py
View file

@ -5664,13 +5664,15 @@ TRANSLATIONS = {
},
}
def get_lang(lang_cookie: str = None, accept_language: str = None) -> str:
"""Get language from cookie or Accept-Language header.
def get_lang(lang_cookie: str = None, accept_language: str = None, lang_query: str = None) -> str:
"""Get language from query param, cookie, or Accept-Language header.
Note: ?lang=XX query param is handled by LanguageMiddleware in app.py,
which sets the cookie and redirects. See Easter egg #0.
Priority: ?lang=XX query > cookie > Accept-Language > "en"
"""
# Cookie takes priority (user's explicit choice via dropdown or ?lang=XX)
# Query param takes highest priority (Easter egg #0: ?lang=XX)
if lang_query and lang_query.lower() in TRANSLATIONS:
return lang_query.lower()
# Cookie next (user's explicit choice via dropdown)
if lang_cookie and lang_cookie in TRANSLATIONS:
return lang_cookie
# Fall back to Accept-Language header

51
serp/server.py
View file

@ -25,7 +25,7 @@ import os
from pathlib import Path
from typing import List, Dict, Any, Optional
from fastapi import Query, HTTPException, BackgroundTasks, Header, Cookie, UploadFile, File, Request
from fastapi import Query, HTTPException, BackgroundTasks, Header, Cookie, UploadFile, File, Request, Depends
from fastapi.responses import FileResponse, HTMLResponse, JSONResponse, RedirectResponse, Response, StreamingResponse
from sqlalchemy import text
import uvicorn
@ -38,8 +38,8 @@ from neopig.live import get_live_queue
# Import from serp package (relative imports since we're inside the package)
import importlib
app_config = importlib.import_module('serp.app') # Module for mutable config (DB_PATH, VAULT_PATH)
from .app import app, templates, CRAWL_DISABLED, IMPORT_MODE
from .i18n import TRANSLATIONS, LANG_NAMES, get_lang, t, inject_i18n, NAV_HTML, SEARCH_BOX_HTML
from .app import app, templates, CRAWL_DISABLED, IMPORT_MODE, get_language
from .i18n import TRANSLATIONS, LANG_NAMES, t, inject_i18n, NAV_HTML, SEARCH_BOX_HTML
from .models import CrawlRequest
from .archive import ArchiveDB
from .logging import start_job_logging, stop_job_logging, get_job_logs
@ -118,9 +118,8 @@ async def shutdown_event():
@app.get("/", response_class=HTMLResponse)
async def index(request: Request, lang: str = Cookie(None), accept_language: str = Header(None)):
async def index(request: Request, language: str = Depends(get_language)):
"""Simple search UI. Use ?lang=XX to force language (e.g., ?lang=ka for Georgian)."""
language = get_lang(lang, accept_language)
t = TRANSLATIONS.get(language, TRANSLATIONS["en"])
if templates:
@ -137,9 +136,8 @@ async def index(request: Request, lang: str = Cookie(None), accept_language: str
@app.get("/crawl", response_class=HTMLResponse)
async def crawl_page(request: Request, lang: str = Cookie(None), accept_language: str = Header(None)):
async def crawl_page(request: Request, language: str = Depends(get_language)):
"""Crawler command page. Use ?lang=XX to force language."""
language = get_lang(lang, accept_language)
t = TRANSLATIONS.get(language, TRANSLATIONS["en"])
# Use Jinja2 template if available, otherwise fall back to inline HTML
@ -163,9 +161,8 @@ async def crawl_page(request: Request, lang: str = Cookie(None), accept_language
@app.get("/live", response_class=HTMLResponse)
async def live_page(request: Request, domain: str = Query(None), lang: str = Cookie(None), accept_language: str = Header(None)):
async def live_page(request: Request, domain: str = Query(None), language: str = Depends(get_language)):
"""Live feed page - watch images appear as they're crawled."""
language = get_lang(lang, accept_language)
t = TRANSLATIONS.get(language, TRANSLATIONS["en"])
if templates:
@ -183,9 +180,8 @@ async def live_page(request: Request, domain: str = Query(None), lang: str = Coo
@app.get("/view/{md5_hash}", response_class=HTMLResponse)
async def view_media_page(request: Request, md5_hash: str, noai: bool = Query(False), lang: str = Cookie(None), accept_language: str = Header(None)):
async def view_media_page(request: Request, md5_hash: str, noai: bool = Query(False), language: str = Depends(get_language)):
"""Detail view page for a single media item. Use ?lang=XX to force language."""
language = get_lang(lang, accept_language)
import html as html_module
import re
from urllib.parse import quote, urljoin
@ -474,31 +470,27 @@ async def view_media_page(request: Request, md5_hash: str, noai: bool = Query(Fa
@app.get("/page/{uri_hash}", response_class=HTMLResponse)
async def view_page_by_hash(request: Request,
uri_hash: str,
noai: bool = Query(False, description="Disable AI assistant"),
lang: str = Cookie(None),
accept_language: str = Header(None),
):
"""View an archived page by URI hash."""
@app.get("/page/{uri_hash}")
async def view_page_by_hash(uri_hash: str, noai: bool = Query(False, description="Disable AI assistant")):
"""View an archived page by URI hash - redirects to /page/view."""
page = await db.get_page_by_hash(uri_hash)
if not page:
raise HTTPException(status_code=404, detail="Page not found")
# Redirect to the URI-based view (reuses same logic)
return await view_page(request=request, uri=page['uri'], noai=noai, lang=lang, accept_language=accept_language)
# Redirect to the URI-based view (language handled by Depends in view_page)
from urllib.parse import quote
redirect_url = f"/page/view?uri={quote(page['uri'], safe='')}"
if noai:
redirect_url += "&noai=true"
return RedirectResponse(url=redirect_url, status_code=302)
@app.get("/page/view", response_class=HTMLResponse)
async def view_page(request: Request,
uri: str = Query(..., description="Page URI to view"),
noai: bool = Query(False, description="Disable AI assistant"),
lang: str = Query(None),
lang_cookie: str = Cookie(None, alias="lang"),
accept_language: str = Header(None),
language: str = Depends(get_language),
):
"""View an archived page with markdown and screenshot. Use ?lang=XX to force language."""
language = get_lang(lang, accept_language)
import html as html_module
import re
from urllib.parse import urljoin, quote
@ -755,9 +747,8 @@ async def phantom_export(domain: str = Query(None, description="Filter by domain
@app.get("/phantom", response_class=HTMLResponse)
async def phantom_page(request: Request, lang: str = Cookie(None), accept_language: str = Header(None)):
async def phantom_page(request: Request, language: str = Depends(get_language)):
"""Phantom site export UI."""
language = get_lang(lang, accept_language)
t = TRANSLATIONS.get(language, TRANSLATIONS["en"])
domains = await db.get_domains_with_pages()
@ -776,9 +767,8 @@ async def phantom_page(request: Request, lang: str = Cookie(None), accept_langua
@app.get("/about", response_class=HTMLResponse)
async def about_page(request: Request, lang: str = Cookie(None), accept_language: str = Header(None)):
async def about_page(request: Request, language: str = Depends(get_language)):
"""About neopig - the story of pig.py's evolution."""
language = get_lang(lang, accept_language)
t = TRANSLATIONS.get(language, TRANSLATIONS["en"])
if templates:
@ -1561,11 +1551,10 @@ async def import_status():
@app.get("/import", response_class=HTMLResponse)
async def import_page(request: Request, lang: str = Cookie(None), accept_language: str = Header(None)):
async def import_page(request: Request, language: str = Depends(get_language)):
"""Import page for uploading archives."""
if not IMPORT_MODE:
raise HTTPException(status_code=403, detail="Import mode not enabled (NEOPIG_IMPORT=1)")
language = get_lang(lang, accept_language)
t = TRANSLATIONS.get(language, TRANSLATIONS["en"])
if templates: