#!/usr/bin/env python3
import html
import json
import sqlite3
import threading
import time
from pathlib import Path
from urllib.parse import parse_qs, urlparse
from wsgiref.simple_server import make_server
from shorts_pipeline import AGENTS, evaluate_ratings, readiness
DB_PATH = Path('/root/workspace/shorts_data/shorts.db')
def init_db(path=DB_PATH):
path = Path(path); path.parent.mkdir(parents=True, exist_ok=True)
with sqlite3.connect(path) as db:
db.executescript('''
CREATE TABLE IF NOT EXISTS settings (key TEXT PRIMARY KEY, value TEXT NOT NULL);
CREATE TABLE IF NOT EXISTS candidates (
id INTEGER PRIMARY KEY AUTOINCREMENT, source_url TEXT, title TEXT NOT NULL,
clip_path TEXT, status TEXT NOT NULL DEFAULT 'pending', score REAL,
ratings_json TEXT NOT NULL DEFAULT '[]', warnings_json TEXT NOT NULL DEFAULT '[]',
created_at INTEGER NOT NULL, published_at INTEGER
);
INSERT OR IGNORE INTO settings(key,value) VALUES ('threshold','85');
INSERT OR IGNORE INTO settings(key,value) VALUES ('publishing_enabled','0');
INSERT OR IGNORE INTO settings(key,value) VALUES ('source_url','');
''')
def score_candidate(scores, threshold=85, safety_score=None):
scores = [float(s) for s in scores]
if not scores: return False
safety = scores[-1] if safety_score is None else float(safety_score)
return sum(scores) / len(scores) >= threshold and sorted(scores)[len(scores)//2] >= threshold - 3 and safety >= 90 and sum(s >= 80 for s in scores) >= 4
def _settings(db):
return {k:v for k,v in db.execute('SELECT key,value FROM settings')}
def _json(start, status=200, body=None):
data = json.dumps(body if body is not None else {}).encode()
start({200:'200 OK',201:'201 Created',404:'404 Not Found'}.get(status, f'{status} Error'), [('Content-Type','application/json'),('Content-Length',str(len(data)))])
return [data]
class Response:
def __init__(self, status, body): self.status_code=status; self._body=body
def get_json(self): return json.loads(self._body)
class TestClient:
def __init__(self, app): self.app=app
def get(self, path):
result=[]; headers=[]
body=b''.join(self.app._wsgi({'REQUEST_METHOD':'GET','PATH_INFO':path,'wsgi.input':None}, lambda s,h: (result.append(s), headers.extend(h))))
return Response(int(result[0].split()[0]), body)
class App:
def __init__(self, db): self.db=Path(db); self.testing=False
def test_client(self): return TestClient(self)
def _wsgi(self, environ, start):
path=environ.get('PATH_INFO','/')
with sqlite3.connect(self.db) as db:
if path.startswith('/api/preview/'):
row = db.execute('SELECT clip_path FROM candidates WHERE id=?', (path.rsplit('/',1)[-1],)).fetchone()
if not row or not row[0]: return _json(start,404,{'error':'No preview'})
target = Path(row[0]).resolve()
if not target.is_relative_to(Path('/root/workspace/shorts_data/media')) or not target.is_file(): return _json(start,404,{'error':'No preview'})
data = target.read_bytes()
start('200 OK',[('Content-Type','video/mp4'),('Content-Length',str(len(data)))])
return [data]
if path == '/api/health': return _json(start, body={'ok':True})
if path == '/api/readiness': return _json(start, body=readiness())
if path == '/api/agents': return _json(start, body=[{'id': key, 'mission': mission} for key, mission in AGENTS])
if path == '/api/summary':
s=_settings(db); counts=dict(db.execute('SELECT status,count(*) FROM candidates GROUP BY status'))
return _json(start, body={'threshold':float(s['threshold']),'publishing_enabled':s['publishing_enabled']=='1','source_url':s['source_url'],'counts':counts,'worker_status':s.get('worker_status','unknown'),'scan_started':s.get('scan_started'),'last_scan':s.get('last_scan'),'last_error':s.get('last_error',''),'last_result':json.loads(s.get('last_result','{}'))})
if path == '/api/candidates':
rows=[dict(zip(['id','source_url','title','clip_path','status','score','ratings','warnings','created_at'],r)) for r in db.execute('SELECT id,source_url,title,clip_path,status,score,ratings_json,warnings_json,created_at FROM candidates ORDER BY created_at DESC')]
for r in rows: r['ratings']=json.loads(r.pop('ratings')); r['warnings']=json.loads(r.pop('warnings'))
return _json(start, body=rows)
if path == '/api/settings' and environ.get('REQUEST_METHOD')=='POST':
length=int(environ.get('CONTENT_LENGTH') or 0); payload=json.loads(environ['wsgi.input'].read(length) or b'{}')
for k in ('threshold','source_url'):
if k in payload: db.execute('INSERT OR REPLACE INTO settings(key,value) VALUES (?,?)',(k,str(payload[k])))
if 'publishing_enabled' in payload: db.execute('INSERT OR REPLACE INTO settings(key,value) VALUES (?,?)',('publishing_enabled','1' if payload['publishing_enabled'] else '0'))
db.commit(); return _json(start, body={'ok':True})
if path == '/api/candidates' and environ.get('REQUEST_METHOD')=='POST':
length=int(environ.get('CONTENT_LENGTH') or 0); p=json.loads(environ['wsgi.input'].read(length) or b'{}')
ratings=p.get('ratings',[]); gate=evaluate_ratings(ratings, float(_settings(db)['threshold']), p.get('warnings',[])) if ratings else None
score=gate.average if gate else None
status='approved' if gate and gate.approved else 'pending'
db.execute('INSERT INTO candidates(source_url,title,clip_path,status,score,ratings_json,warnings_json,created_at) VALUES (?,?,?,?,?,?,?,?)',(p.get('source_url',''),p.get('title','Untitled'),p.get('clip_path',''),status,score,json.dumps(ratings),json.dumps(p.get('warnings',[])),int(time.time())))
db.commit(); return _json(start, status=201, body={'ok':True,'status':status,'score':score})
if path == '/': return self._page(start, db)
return _json(start,404,{'error':'not found'})
def _page(self,start,db):
data=_settings(db); start('200 OK',[('Content-Type','text/html; charset=utf-8')])
return [Path('/root/workspace/shorts_ui.html').read_text().replace('__THRESHOLD__',html.escape(data['threshold'])).replace('__SOURCE__',html.escape(data['source_url'])).encode()]
def __call__(self,environ,start): return self._wsgi(environ,start)
def create_app(db=DB_PATH): init_db(db); return App(db)
PAGE='''Shorts ControlShorts Control
ECC-inspired scoring dashboard. Public publishing is disabled until platform OAuth and source rights are configured.
'''
if __name__=='__main__':
app=create_app(); print('shorts dashboard listening on 127.0.0.1:8790', flush=True)
make_server('127.0.0.1',8790,app).serve_forever()