from __future__ import annotations import threading from typing import Any, Dict, List, Optional from fastapi import APIRouter, Depends, Header, HTTPException, Path, Query from fastapi.responses import PlainTextResponse from pydantic import BaseModel, Field from app.config import get_settings from app.services.url_shortener_service import ( AliasTakenError, AuthenticationError, AuthorizationError, LinkExpiredError, LinkNotFoundError, PlanLimitExceededError, URLShortenerError, URLShortenerService, ValidationError, ) _settings = get_settings() router = APIRouter() _service_lock = threading.Lock() _shared_service: Optional[URLShortenerService] = None def _get_service() -> URLShortenerService: global _shared_service if _shared_service is None: with _service_lock: if _shared_service is None: _shared_service = URLShortenerService() return _shared_service # --------------------------------------------------------------------------- # Request / Response models # --------------------------------------------------------------------------- class CreateOwnerRequest(BaseModel): name: str = Field(..., min_length=1, max_length=200) plan: Optional[str] = Field("free", description="free, pro, or enterprise") class CreateOwnerResponse(BaseModel): success: bool = True owner_id: str api_key: str plan: str class ShortenRequest(BaseModel): long_url: str = Field(..., description="Destination URL (http/https only)") custom_alias: Optional[str] = Field(None, description="Custom short code (3-32 chars, letters/digits/-/_)") expires_in_days: Optional[int] = Field(None, ge=1) max_clicks: Optional[int] = Field(None, ge=1) tags: Optional[List[str]] = Field(None, description="Tags (max 20)") note: Optional[str] = Field(None, description="Max 2000 chars") password: Optional[str] = Field(None, description="Password-protect the link") utm_source: Optional[str] = Field(None) utm_medium: Optional[str] = Field(None) utm_campaign: Optional[str] = Field(None) campaign_id: Optional[str] = Field(None, description="Campaign to group this link under") custom_domain: Optional[str] = Field(None, description="Branded domain override") fallback_url: Optional[str] = Field(None, description="Redirect here when link is expired/deactivated") webhook_url: Optional[str] = Field(None, description="URL to POST click events to (Enterprise)") geo_targeting: Optional[Dict[str, Any]] = Field(None, description='{"countries": {...}, "devices": {...}}') class ShortLinkResponse(BaseModel): success: bool = True short_code: str long_url: str owner_id: str created_at: str expires_at: Optional[str] = None max_clicks: Optional[int] = None click_count: int = 0 is_active: bool = True tags: List[str] = [] note: Optional[str] = None has_password: bool = False short_url: Optional[str] = None qr_image_url: Optional[str] = None campaign_id: Optional[str] = None custom_domain: Optional[str] = None fallback_url: Optional[str] = None webhook_url: Optional[str] = None class UpdateLinkRequest(BaseModel): long_url: Optional[str] = None note: Optional[str] = None tags: Optional[List[str]] = None max_clicks: Optional[int] = None password: Optional[str] = None webhook_url: Optional[str] = None fallback_url: Optional[str] = None campaign_id: Optional[str] = None custom_domain: Optional[str] = None geo_targeting: Optional[Dict[str, Any]] = None class ListLinksResponse(BaseModel): success: bool = True links: List[ShortLinkResponse] class StatsResponse(BaseModel): success: bool = True short_code: str long_url: str click_count: int created_at: str last_accessed_at: Optional[str] = None is_active: bool expires_at: Optional[str] = None max_clicks: Optional[int] = None has_password: bool = False utm: Optional[Dict[str, Any]] = None analytics: Optional[Dict[str, Any]] = None campaign_id: Optional[str] = None custom_domain: Optional[str] = None fallback_url: Optional[str] = None webhook_url: Optional[str] = None class ClickDetail(BaseModel): id: int referrer: Optional[str] = None user_agent: Optional[str] = None ip_address: Optional[str] = None browser: Optional[str] = None device: Optional[str] = None os: Optional[str] = None clicked_at: str class ClickDetailsResponse(BaseModel): success: bool = True clicks: List[ClickDetail] class QRCodeResponse(BaseModel): success: bool = True short_code: str short_url: str qr_image_url: str class ErrorResponse(BaseModel): success: bool = False error: str class BulkShortenRequest(BaseModel): items: List[Dict[str, Any]] = Field(..., description="Array of shorten requests (max 100)") class BulkShortenResponse(BaseModel): success: bool = True results: List[Dict[str, Any]] class OwnerSummaryResponse(BaseModel): success: bool = True owner_id: str name: str plan: str total_links: int active_links: int total_clicks: int created_at: str class CampaignCreateRequest(BaseModel): name: str = Field(..., min_length=1, max_length=200) description: Optional[str] = Field(None, max_length=2000) class CampaignResponse(BaseModel): success: bool = True campaign_id: str name: str description: Optional[str] = None created_at: Optional[str] = None is_active: Optional[bool] = None link_count: Optional[int] = 0 total_clicks: Optional[int] = 0 analytics: Optional[Dict[str, Any]] = None class CampaignListResponse(BaseModel): success: bool = True campaigns: List[CampaignResponse] class HealthCheckResponse(BaseModel): success: bool = True short_code: str long_url: str reachable: bool status_code: Optional[int] = None error: Optional[str] = None # --------------------------------------------------------------------------- # Helpers # --------------------------------------------------------------------------- def _auth_owner(x_api_key: str = Header(..., alias="X-API-Key")) -> str: service = _get_service() try: return service.authenticate(x_api_key) except AuthenticationError: raise HTTPException(status_code=401, detail={"success": False, "error": "Invalid API key"}) def _build_link_response(link, service) -> ShortLinkResponse: return ShortLinkResponse( short_code=link.short_code, long_url=link.long_url, owner_id=link.owner_id, created_at=link.created_at.isoformat(), expires_at=link.expires_at.isoformat() if link.expires_at else None, max_clicks=link.max_clicks, click_count=link.click_count, is_active=link.is_active, tags=link.tags, note=link.note, has_password=bool(link.password_hash) if hasattr(link, 'password_hash') else False, campaign_id=link.campaign_id, custom_domain=link.custom_domain, fallback_url=link.fallback_url, webhook_url=link.webhook_url, ) # --------------------------------------------------------------------------- # Owner / Plan # --------------------------------------------------------------------------- @router.post( "/url-shortener/owners", response_model=CreateOwnerResponse, summary="Create a new owner (tenant) and issue API key", ) def create_owner(body: CreateOwnerRequest): service = _get_service() result = service.create_owner(body.name, body.plan or "free") return CreateOwnerResponse( owner_id=result["owner_id"], api_key=result["api_key"], plan=body.plan or "free", ) @router.patch( "/url-shortener/plan", summary="Update owner's subscription plan", ) def update_plan(plan: str = Query(..., description="free, pro, or enterprise"), owner_id: str = Depends(_auth_owner)): service = _get_service() try: return {"success": True, **service.update_owner_plan(owner_id, plan)} except ValidationError as e: raise HTTPException(status_code=400, detail={"success": False, "error": str(e)}) @router.get( "/url-shortener/summary", response_model=OwnerSummaryResponse, summary="Get owner dashboard summary", ) def owner_summary(owner_id: str = Depends(_auth_owner)): service = _get_service() data = service.owner_summary(owner_id) if not data: raise HTTPException(status_code=404, detail={"success": False, "error": "Owner not found"}) return OwnerSummaryResponse(**data) # --------------------------------------------------------------------------- # Short links CRUD # --------------------------------------------------------------------------- @router.post( "/url-shortener/shorten", response_model=ShortLinkResponse, summary="Create a short link", ) def shorten_url(body: ShortenRequest, owner_id: str = Depends(_auth_owner)): service = _get_service() try: link = service.shorten( long_url=body.long_url, owner_id=owner_id, custom_alias=body.custom_alias, expires_in_days=body.expires_in_days, max_clicks=body.max_clicks, tags=body.tags, note=body.note, password=body.password, utm_source=body.utm_source, utm_medium=body.utm_medium, utm_campaign=body.utm_campaign, campaign_id=body.campaign_id, custom_domain=body.custom_domain, fallback_url=body.fallback_url, webhook_url=body.webhook_url, geo_targeting=body.geo_targeting, ) from app.services.url_shortener_service import SRV_BASE_URL resp = _build_link_response(link, service) short_url = f"{SRV_BASE_URL}/url-shortener/{link.short_code}" if link.custom_domain: short_url = f"https://{link.custom_domain}/{link.short_code}" resp.short_url = short_url resp.qr_image_url = f"https://api.qrserver.com/v1/create-qr-code/?size=300x300&data={short_url}" return resp except ValidationError as e: raise HTTPException(status_code=400, detail={"success": False, "error": str(e)}) except AuthenticationError as e: raise HTTPException(status_code=401, detail={"success": False, "error": str(e)}) except AliasTakenError as e: raise HTTPException(status_code=409, detail={"success": False, "error": str(e)}) except PlanLimitExceededError as e: raise HTTPException(status_code=402, detail={"success": False, "error": str(e), "code": "plan_limit"}) except URLShortenerError as e: raise HTTPException(status_code=500, detail={"success": False, "error": str(e)}) @router.post( "/url-shortener/bulk", response_model=BulkShortenResponse, summary="Bulk create short links", ) def bulk_shorten(body: BulkShortenRequest, owner_id: str = Depends(_auth_owner)): if len(body.items) > 100: raise HTTPException(status_code=400, detail={"success": False, "error": "Bulk limit is 100 items per request"}) service = _get_service() results = service.bulk_shorten(owner_id, body.items) return BulkShortenResponse(results=results, ) @router.get( "/url-shortener/links", response_model=ListLinksResponse, summary="List all links for the authenticated owner", ) def list_links(owner_id: str = Depends(_auth_owner)): service = _get_service() links = service.list_links(owner_id) items = [_build_link_response(link, service) for link in links] return ListLinksResponse(links=items, ) @router.get( "/url-shortener/export", summary="Export all links as CSV", ) def export_links_csv(owner_id: str = Depends(_auth_owner)): service = _get_service() csv_data = service.export_links(owner_id) return PlainTextResponse(csv_data, media_type="text/csv", headers={"Content-Disposition": "attachment; filename=links_export.csv"}) @router.post( "/url-shortener/campaigns", response_model=CampaignResponse, summary="Create a campaign to group links", ) def create_campaign(body: CampaignCreateRequest, owner_id: str = Depends(_auth_owner)): service = _get_service() try: result = service.create_campaign(owner_id, body.name, body.description) return CampaignResponse( campaign_id=result["campaign_id"], name=result["name"], description=result.get("description"), ) except PlanLimitExceededError as e: raise HTTPException(status_code=402, detail={"success": False, "error": str(e), "code": "plan_limit"}) except ValidationError as e: raise HTTPException(status_code=400, detail={"success": False, "error": str(e)}) @router.get( "/url-shortener/campaigns", response_model=CampaignListResponse, summary="List all campaigns for the authenticated owner", ) def list_campaigns(owner_id: str = Depends(_auth_owner)): service = _get_service() campaigns = service.list_campaigns(owner_id) items = [CampaignResponse(**c, ) for c in campaigns] return CampaignListResponse(campaigns=items, ) @router.get( "/url-shortener/campaigns/{campaign_id}", response_model=CampaignResponse, summary="Get campaign details with analytics", ) def get_campaign( campaign_id: str = Path(...), owner_id: str = Depends(_auth_owner), ): service = _get_service() try: data = service.get_campaign(campaign_id, owner_id) return CampaignResponse(**data, ) except LinkNotFoundError as e: raise HTTPException(status_code=404, detail={"success": False, "error": str(e)}) except AuthorizationError as e: raise HTTPException(status_code=403, detail={"success": False, "error": str(e)}) @router.get( "/url-shortener/campaigns/{campaign_id}/links", response_model=ListLinksResponse, summary="List all links in a campaign", ) def list_campaign_links( campaign_id: str = Path(...), owner_id: str = Depends(_auth_owner), ): service = _get_service() try: links = service.list_campaign_links(campaign_id, owner_id) items = [_build_link_response(link, service) for link in links] return ListLinksResponse(links=items, ) except LinkNotFoundError as e: raise HTTPException(status_code=404, detail={"success": False, "error": str(e)}) except AuthorizationError as e: raise HTTPException(status_code=403, detail={"success": False, "error": str(e)}) @router.delete( "/url-shortener/campaigns/{campaign_id}", summary="Deactivate a campaign", ) def deactivate_campaign( campaign_id: str = Path(...), owner_id: str = Depends(_auth_owner), ): service = _get_service() try: service.deactivate_campaign(campaign_id, owner_id) return {"success": True, "message": f"Campaign '{campaign_id}' deactivated"} except LinkNotFoundError as e: raise HTTPException(status_code=404, detail={"success": False, "error": str(e)}) except AuthorizationError as e: raise HTTPException(status_code=403, detail={"success": False, "error": str(e)}) # --------------------------------------------------------------------------- # Short link path-parameter routes (must come after static routes) # --------------------------------------------------------------------------- @router.get( "/url-shortener/{short_code}", summary="Resolve a short code (redirect to long URL)", ) def resolve_short_code( short_code: str = Path(...), x_link_password: Optional[str] = Header(None, alias="X-Link-Password"), referer: Optional[str] = Header(None, alias="Referer"), user_agent: Optional[str] = Header(None, alias="User-Agent"), x_forwarded_for: Optional[str] = Header(None, alias="X-Forwarded-For"), ): service = _get_service() from fastapi.responses import RedirectResponse ip = x_forwarded_for.split(",")[0].strip() if x_forwarded_for else None try: if x_link_password: long_url = service.resolve_with_password(short_code, x_link_password, referer, user_agent, ip) else: long_url = service.resolve(short_code, referer, user_agent, ip) return RedirectResponse(url=long_url, status_code=302) except LinkNotFoundError as e: raise HTTPException(status_code=404, detail={"success": False, "error": str(e)}) except LinkExpiredError as e: raise HTTPException(status_code=410, detail={"success": False, "error": str(e)}) except AuthorizationError as e: raise HTTPException(status_code=403, detail={"success": False, "error": str(e)}) @router.patch( "/url-shortener/{short_code}", response_model=ShortLinkResponse, summary="Update a short link (change URL, tags, webhook, etc.)", ) def update_short_link( body: UpdateLinkRequest, short_code: str = Path(...), owner_id: str = Depends(_auth_owner), ): service = _get_service() try: updates = {k: v for k, v in body.model_dump().items() if v is not None} if not updates: raise HTTPException(status_code=400, detail={"success": False, "error": "No fields to update"}) link = service.update_link(short_code, owner_id, **updates) return _build_link_response(link, service) except (ValidationError, AuthenticationError) as e: raise HTTPException(status_code=400, detail={"success": False, "error": str(e)}) except LinkNotFoundError as e: raise HTTPException(status_code=404, detail={"success": False, "error": str(e)}) except AuthorizationError as e: raise HTTPException(status_code=403, detail={"success": False, "error": str(e)}) @router.delete( "/url-shortener/{short_code}", summary="Deactivate a short link", ) def deactivate_link( short_code: str = Path(...), owner_id: str = Depends(_auth_owner), ): service = _get_service() try: service.deactivate_link(short_code, owner_id) return {"success": True, "message": f"Link '{short_code}' deactivated"} except LinkNotFoundError as e: raise HTTPException(status_code=404, detail={"success": False, "error": str(e)}) except AuthorizationError as e: raise HTTPException(status_code=403, detail={"success": False, "error": str(e)}) @router.get( "/url-shortener/{short_code}/stats", response_model=StatsResponse, summary="Get click statistics for a link (browsers, devices, referrers, etc.)", ) def get_link_stats( short_code: str = Path(...), owner_id: str = Depends(_auth_owner), ): service = _get_service() try: stats = service.get_link_stats(short_code, owner_id) return StatsResponse( short_code=stats["short_code"], long_url=stats["long_url"], click_count=stats["click_count"], created_at=stats["created_at"], last_accessed_at=stats["last_accessed_at"], is_active=stats["is_active"], expires_at=stats["expires_at"], max_clicks=stats["max_clicks"], has_password=stats["has_password"], utm=stats["utm"], analytics=stats["analytics"], campaign_id=stats.get("campaign_id"), custom_domain=stats.get("custom_domain"), fallback_url=stats.get("fallback_url"), webhook_url=stats.get("webhook_url"), ) except LinkNotFoundError as e: raise HTTPException(status_code=404, detail={"success": False, "error": str(e)}) except AuthorizationError as e: raise HTTPException(status_code=403, detail={"success": False, "error": str(e)}) @router.get( "/url-shortener/{short_code}/clicks", response_model=ClickDetailsResponse, summary="Get detailed click log for a link", ) def get_click_details( short_code: str = Path(...), limit: int = Query(100, ge=1, le=1000), owner_id: str = Depends(_auth_owner), ): service = _get_service() try: clicks = service.get_click_details(short_code, owner_id, limit) return ClickDetailsResponse(clicks=[ClickDetail(**c) for c in clicks], ) except LinkNotFoundError as e: raise HTTPException(status_code=404, detail={"success": False, "error": str(e)}) except AuthorizationError as e: raise HTTPException(status_code=403, detail={"success": False, "error": str(e)}) @router.get( "/url-shortener/{short_code}/qr", response_model=QRCodeResponse, summary="Get QR code for a short link", ) def get_qr_code( short_code: str = Path(...), ): service = _get_service() try: data = service.get_qr_code(short_code) return QRCodeResponse(**data, ) except LinkNotFoundError as e: raise HTTPException(status_code=404, detail={"success": False, "error": str(e)}) @router.get( "/url-shortener/{short_code}/health", response_model=HealthCheckResponse, summary="Check if the destination URL is reachable", ) def link_health( short_code: str = Path(...), owner_id: str = Depends(_auth_owner), ): service = _get_service() try: result = service.check_link_health(short_code, owner_id) return HealthCheckResponse(**result, ) except LinkNotFoundError as e: raise HTTPException(status_code=404, detail={"success": False, "error": str(e)}) except AuthorizationError as e: raise HTTPException(status_code=403, detail={"success": False, "error": str(e)}) @router.get( "/url-shortener/{short_code}/clicks/export", summary="Export click data as CSV", ) def export_clicks_csv( short_code: str = Path(...), owner_id: str = Depends(_auth_owner), ): service = _get_service() try: csv_data = service.export_clicks(short_code, owner_id) return PlainTextResponse(csv_data, media_type="text/csv", headers={"Content-Disposition": f"attachment; filename=clicks_{short_code}.csv"}) except LinkNotFoundError as e: raise HTTPException(status_code=404, detail={"success": False, "error": str(e)}) except AuthorizationError as e: raise HTTPException(status_code=403, detail={"success": False, "error": str(e)})