"""
Tikker API with C Tools Integration
FastAPI endpoints that call C tools for statistics and report generation.
Maintains 100% backwards compatibility with original API interface.
"""
from fastapi import FastAPI, HTTPException, BackgroundTasks, Query
from fastapi.responses import FileResponse, HTMLResponse
from pydantic import BaseModel
from typing import List, Dict, Any, Optional
import logging
import os
from pathlib import Path
from c_tools_wrapper import CToolsWrapper, ToolError
# Configure logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
# Initialize FastAPI app
app = FastAPI(
title="Tikker API",
description="Enterprise keystroke analytics API with C backend",
version="2.0.0"
)
# Initialize C tools wrapper
try:
tools = CToolsWrapper(
tools_dir=os.getenv("TOOLS_DIR", "./build/bin"),
db_path=os.getenv("DB_PATH", "tikker.db")
)
except Exception as e:
logger.error(f"Failed to initialize C tools: {e}")
tools = None
# Pydantic models
class DailyStats(BaseModel):
presses: int
releases: int
repeats: int
total: int
class WordStat(BaseModel):
rank: int
word: str
count: int
percentage: float
class DecoderRequest(BaseModel):
input_file: str
output_file: str
verbose: bool = False
class ReportRequest(BaseModel):
output_file: str = "report.html"
input_dir: str = "logs_plain"
title: str = "Tikker Activity Report"
# Health check endpoint
@app.get("/health")
async def health_check() -> Dict[str, Any]:
"""
Check API and C tools health status.
Returns:
Health status and tool information
"""
if not tools:
raise HTTPException(status_code=503, detail="C tools not initialized")
return tools.health_check()
# Statistics endpoints
@app.get("/api/stats/daily", response_model=DailyStats)
async def get_daily_stats() -> DailyStats:
"""
Get daily keystroke statistics.
Returns:
Daily statistics (presses, releases, repeats, total)
"""
if not tools:
raise HTTPException(status_code=503, detail="C tools not available")
try:
stats = tools.get_daily_stats()
return DailyStats(**stats)
except ToolError as e:
raise HTTPException(status_code=500, detail=str(e))
@app.get("/api/stats/hourly")
async def get_hourly_stats(date: str = Query(..., description="Date in YYYY-MM-DD format")) -> Dict[str, Any]:
"""
Get hourly keystroke statistics for a specific date.
Args:
date: Date in YYYY-MM-DD format
Returns:
Hourly statistics
"""
if not tools:
raise HTTPException(status_code=503, detail="C tools not available")
try:
return tools.get_hourly_stats(date)
except ToolError as e:
raise HTTPException(status_code=500, detail=str(e))
@app.get("/api/stats/weekly")
async def get_weekly_stats() -> Dict[str, Any]:
"""
Get weekly keystroke statistics.
Returns:
Weekly statistics breakdown
"""
if not tools:
raise HTTPException(status_code=503, detail="C tools not available")
try:
return tools.get_weekly_stats()
except ToolError as e:
raise HTTPException(status_code=500, detail=str(e))
@app.get("/api/stats/weekday")
async def get_weekday_stats() -> Dict[str, Any]:
"""
Get weekday comparison statistics.
Returns:
Statistics grouped by day of week
"""
if not tools:
raise HTTPException(status_code=503, detail="C tools not available")
try:
return tools.get_weekday_stats()
except ToolError as e:
raise HTTPException(status_code=500, detail=str(e))
# Word analysis endpoints
@app.get("/api/words/top", response_model=List[WordStat])
async def get_top_words(limit: int = Query(10, ge=1, le=100, description="Number of words to return")) -> List[WordStat]:
"""
Get top N most popular words.
Args:
limit: Number of words to return (1-100)
Returns:
List of words with frequency and rank
"""
if not tools:
raise HTTPException(status_code=503, detail="C tools not available")
try:
words = tools.get_top_words(limit)
return [WordStat(**w) for w in words]
except ToolError as e:
raise HTTPException(status_code=500, detail=str(e))
@app.get("/api/words/find")
async def find_word(word: str = Query(..., description="Word to search for")) -> Dict[str, Any]:
"""
Find statistics for a specific word.
Args:
word: Word to search for
Returns:
Word frequency, rank, and statistics
"""
if not tools:
raise HTTPException(status_code=503, detail="C tools not available")
try:
return tools.find_word(word)
except ToolError as e:
raise HTTPException(status_code=500, detail=str(e))
# Indexing endpoints
@app.post("/api/index")
async def build_index(dir_path: str = Query("logs_plain", description="Directory to index")) -> Dict[str, Any]:
"""
Build word index from text files.
Args:
dir_path: Directory containing text files
Returns:
Indexing results and statistics
"""
if not tools:
raise HTTPException(status_code=503, detail="C tools not available")
try:
return tools.index_directory(dir_path)
except ToolError as e:
raise HTTPException(status_code=500, detail=str(e))
# Decoding endpoints
@app.post("/api/decode")
async def decode_file(request: DecoderRequest, background_tasks: BackgroundTasks) -> Dict[str, Any]:
"""
Decode keystroke token file to readable text.
Args:
request: Decoder request with input/output paths
background_tasks: Background task runner
Returns:
Decoding result
"""
if not tools:
raise HTTPException(status_code=503, detail="C tools not available")
try:
result = tools.decode_file(request.input_file, request.output_file, request.verbose)
return result
except ToolError as e:
raise HTTPException(status_code=500, detail=str(e))
# Report generation endpoints
@app.post("/api/report")
async def generate_report(request: ReportRequest, background_tasks: BackgroundTasks) -> Dict[str, Any]:
"""
Generate HTML activity report.
Args:
request: Report configuration
background_tasks: Background task runner
Returns:
Report generation result
"""
if not tools:
raise HTTPException(status_code=503, detail="C tools not available")
try:
result = tools.generate_report(
output_file=request.output_file,
input_dir=request.input_dir,
title=request.title
)
return result
except ToolError as e:
raise HTTPException(status_code=500, detail=str(e))
@app.get("/api/report/{filename}")
async def get_report(filename: str) -> FileResponse:
"""
Download generated report file.
Args:
filename: Report filename (without path)
Returns:
File response with report content
"""
file_path = Path(filename)
# Security check - prevent directory traversal
if ".." in filename or "/" in filename:
raise HTTPException(status_code=400, detail="Invalid filename")
if not file_path.exists():
raise HTTPException(status_code=404, detail="Report not found")
return FileResponse(path=file_path, filename=filename, media_type="text/html")
# Root endpoint (for backwards compatibility)
@app.get("/")
async def root() -> Dict[str, Any]:
"""
Root API endpoint.
Returns:
API information
"""
return {
"name": "Tikker API",
"version": "2.0.0",
"status": "running",
"backend": "C tools (libtikker)",
"endpoints": {
"health": "/health",
"stats": "/api/stats/daily, /api/stats/hourly, /api/stats/weekly, /api/stats/weekday",
"words": "/api/words/top, /api/words/find",
"operations": "/api/index, /api/decode, /api/report"
}
}
# Backwards compatibility endpoint
@app.get("/api/all-stats")
async def all_stats() -> Dict[str, Any]:
"""
Get all statistics (backwards compatibility endpoint).
Returns:
Comprehensive statistics
"""
if not tools:
raise HTTPException(status_code=503, detail="C tools not available")
try:
daily = tools.get_daily_stats()
weekly = tools.get_weekly_stats()
top_words = tools.get_top_words(10)
return {
"status": "success",
"daily": daily,
"weekly": weekly,
"top_words": top_words,
"backend": "C"
}
except ToolError as e:
raise HTTPException(status_code=500, detail=str(e))
# Exception handlers
@app.exception_handler(ToolError)
async def tool_error_handler(request, exc):
"""Handle C tool errors."""
logger.error(f"C tool error: {exc}")
return HTTPException(status_code=500, detail=str(exc))
if __name__ == "__main__":
import uvicorn
uvicorn.run(app, host="0.0.0.0", port=8000)
Reference in New Issue View Git Blame Copy Permalink