234 lines
9.3 KiB
Python
234 lines
9.3 KiB
Python
#!/usr/bin/env python3
|
|
"""Minimal SEC filings viewer."""
|
|
|
|
import gzip, json, logging, re, time, urllib.request, urllib.error
|
|
from pathlib import Path
|
|
from flask import Flask, render_template, abort, send_from_directory, request, make_response
|
|
|
|
app = Flask(__name__)
|
|
logging.basicConfig(level=logging.INFO)
|
|
log = logging.getLogger("10k")
|
|
|
|
UA = "10k-viewer/1.0 (test@test.com)"
|
|
CACHE_DIR = Path(__file__).parent / "cache"
|
|
MAX_RETRIES = 1
|
|
PAGE_SIZE = 10
|
|
|
|
FALLBACK = {
|
|
"AAPL": {"cik": "0000320193", "name": "Apple Inc."},
|
|
"MSFT": {"cik": "0000789019", "name": "Microsoft Corp"},
|
|
"GOOGL": {"cik": "0001652044", "name": "Alphabet Inc."},
|
|
"AMZN": {"cik": "0001018724", "name": "Amazon.com Inc."},
|
|
"META": {"cik": "0001326801", "name": "Meta Platforms Inc."},
|
|
"NVDA": {"cik": "0001045810", "name": "NVIDIA Corp"},
|
|
"TSLA": {"cik": "0001318605", "name": "Tesla Inc."},
|
|
"JPM": {"cik": "0000019617", "name": "JPMorgan Chase"},
|
|
"V": {"cik": "0001403161", "name": "Visa Inc."},
|
|
"JNJ": {"cik": "0000200406", "name": "Johnson & Johnson"},
|
|
"WMT": {"cik": "0000104169", "name": "Walmart Inc."},
|
|
"PG": {"cik": "0000080424", "name": "Procter & Gamble"},
|
|
"MA": {"cik": "0001141391", "name": "Mastercard Inc."},
|
|
"HD": {"cik": "0000354950", "name": "Home Depot Inc."},
|
|
"DIS": {"cik": "0001001039", "name": "Walt Disney Co"},
|
|
"NFLX": {"cik": "0001065280", "name": "Netflix Inc."},
|
|
"ADBE": {"cik": "0000796343", "name": "Adobe Inc."},
|
|
"CRM": {"cik": "0001108524", "name": "Salesforce Inc."},
|
|
"INTC": {"cik": "0000050863", "name": "Intel Corp"},
|
|
"AMD": {"cik": "0000002488", "name": "Advanced Micro Devices"},
|
|
"COST": {"cik": "0000909832", "name": "Costco Wholesale"},
|
|
"KO": {"cik": "0000021344", "name": "Coca-Cola Co"},
|
|
"AVGO": {"cik": "0001730168", "name": "Broadcom Inc."},
|
|
"ORCL": {"cik": "0001341439", "name": "Oracle Corp"},
|
|
"LLY": {"cik": "0000059478", "name": "Eli Lilly & Co"},
|
|
"CSCO": {"cik": "0000858877", "name": "Cisco Systems"},
|
|
"MCD": {"cik": "0000063908", "name": "McDonald's Corp"},
|
|
"NKE": {"cik": "0000320187", "name": "Nike Inc."},
|
|
"QCOM": {"cik": "0000804328", "name": "Qualcomm Inc."},
|
|
"TXN": {"cik": "0000010454", "name": "Texas Instruments"},
|
|
"SBUX": {"cik": "0000829224", "name": "Starbucks Corp"},
|
|
"LOW": {"cik": "0000300026", "name": "Lowe's Cos Inc."},
|
|
"PANW": {"cik": "0001742704", "name": "Palo Alto Networks"},
|
|
"NOW": {"cik": "0001590894", "name": "ServiceNow Inc."},
|
|
"AMAT": {"cik": "0000002017", "name": "Applied Materials"},
|
|
"MU": {"cik": "0000727119", "name": "Micron Technology"},
|
|
"ASML": {"cik": "0001647072", "name": "ASML Holding"},
|
|
"SNPS": {"cik": "0000002915", "name": "Synopsys Inc."},
|
|
"CDNS": {"cik": "0000007643", "name": "Cadence Design"},
|
|
"ADI": {"cik": "0000006281", "name": "Analog Devices"},
|
|
"FTNT": {"cik": "0001645674", "name": "Fortinet Inc."},
|
|
"CRWD": {"cik": "0001864275", "name": "CrowdStrike"},
|
|
"ZS": {"cik": "0001783140", "name": "Zscaler Inc."},
|
|
"DDOG": {"cik": "0001779937", "name": "Datadog Inc."},
|
|
"SNOW": {"cik": "0001640185", "name": "Snowflake Inc."},
|
|
"PLTR": {"cik": "0001781878", "name": "Palantir Tech"},
|
|
"COIN": {"cik": "0001679788", "name": "Coinbase Global"},
|
|
"RBLX": {"cik": "0001816155", "name": "Roblox Corp"},
|
|
"SQ": {"cik": "0001594968", "name": "Block Inc."},
|
|
"SHOP": {"cik": "0001594805", "name": "Shopify Inc."},
|
|
"UBER": {"cik": "0001543151", "name": "Uber Technologies"},
|
|
"DASH": {"cik": "0001715041", "name": "DoorDash Inc."},
|
|
"ABNB": {"cik": "0001792947", "name": "Airbnb Inc."},
|
|
"RIVN": {"cik": "0001874178", "name": "Rivian Automotive"},
|
|
"LCID": {"cik": "0001963077", "name": "Lucid Group"},
|
|
"NIO": {"cik": "0001841781", "name": "NIO Inc."},
|
|
"BABA": {"cik": "0001577552", "name": "Alibaba Group"},
|
|
}
|
|
|
|
_tickers = None
|
|
_filings_cache = {}
|
|
|
|
|
|
def fetch_json(url):
|
|
for _ in range(1 + MAX_RETRIES):
|
|
try:
|
|
req = urllib.request.Request(url, headers={"User-Agent": UA})
|
|
with urllib.request.urlopen(req, timeout=15) as r:
|
|
d = r.read()
|
|
if r.headers.get("Content-Encoding") == "gzip":
|
|
d = gzip.decompress(d)
|
|
return json.loads(d.decode("utf-8"))
|
|
except urllib.error.HTTPError as e:
|
|
if e.code in (403, 404, 429):
|
|
raise
|
|
except Exception:
|
|
pass
|
|
raise Exception(f"Failed {url}")
|
|
|
|
|
|
def fetch_filing(url):
|
|
for _ in range(1 + MAX_RETRIES):
|
|
try:
|
|
req = urllib.request.Request(url, headers={"User-Agent": UA})
|
|
with urllib.request.urlopen(req, timeout=30) as r:
|
|
return r.read()
|
|
except Exception as e:
|
|
log.warning(f"Download failed: {e}")
|
|
raise Exception(f"Download failed {url}")
|
|
|
|
|
|
def get_tickers():
|
|
global _tickers
|
|
if _tickers:
|
|
return _tickers
|
|
cf = CACHE_DIR / "tickers.json"
|
|
if cf.exists():
|
|
try:
|
|
_tickers = json.loads(cf.read_text())
|
|
return _tickers
|
|
except Exception:
|
|
pass
|
|
_tickers = FALLBACK
|
|
return _tickers
|
|
|
|
|
|
def parse_filings(subs, cik, forms):
|
|
recent = subs.get("filings", {}).get("recent", {})
|
|
fs = recent.get("form", [])
|
|
ds = recent.get("filingDate", [])
|
|
doc = recent.get("primaryDocument", [])
|
|
acc = recent.get("accessionNumber", [])
|
|
r = []
|
|
for i, f in enumerate(fs):
|
|
if f not in forms:
|
|
continue
|
|
a = acc[i].replace("-", "")
|
|
r.append({
|
|
"form": f,
|
|
"date": ds[i] if i < len(ds) else "?",
|
|
"file": f"{ds[i]}_{a}.html" if i < len(ds) else f"?_{a}.html",
|
|
"url": f"https://www.sec.gov/Archives/edgar/data/{cik}/{a}/{doc[i]}",
|
|
})
|
|
return r
|
|
|
|
|
|
def get_filings(cik, forms):
|
|
key = f"{cik}:{','.join(sorted(forms))}"
|
|
if key in _filings_cache:
|
|
return _filings_cache[key]
|
|
subs = fetch_json(f"https://data.sec.gov/submissions/CIK{cik}.json")
|
|
r = parse_filings(subs, cik, forms)
|
|
_filings_cache[key] = r
|
|
return r
|
|
|
|
|
|
def clean_ereader(html):
|
|
t = html.decode("utf-8", errors="replace")
|
|
t = re.sub(r'<script[^>]*>.*?</script>', '', t, flags=re.DOTALL|re.I)
|
|
t = re.sub(r'<style[^>]*>.*?</style>', '', t, flags=re.DOTALL|re.I)
|
|
t = re.sub(r'\s+style="[^"]*"', '', t, flags=re.I)
|
|
t = re.sub(r'\s+on\w+="[^"]*"', '', t, flags=re.I)
|
|
t = re.sub(r'\s+(class|id)="[^"]*"', '', t, flags=re.I)
|
|
t = re.sub(r'<!--.*?-->', '', t, flags=re.DOTALL)
|
|
s = '<style>body{font-family:Georgia,serif!important;font-size:18px!important;line-height:1.6!important;max-width:700px;margin:0 auto;padding:20px;color:#000;background:#fff}table{border-collapse:collapse;width:100%;margin:10px 0;font-size:14px}th,td{border:1px solid #000;padding:4px 8px;text-align:left}th{font-weight:bold;background:#eee}a{color:#000}h1,h2,h3,h4,h5,h6{font-family:Georgia,serif!important;margin:16px 0 8px}p{margin:8px 0}</style>'
|
|
if '<head>' in t.lower():
|
|
t = t.replace('<head>', f'<head>{s}', 1)
|
|
elif '<body' in t.lower():
|
|
t = t.replace('<body', f'<body{s}', 1)
|
|
else:
|
|
t = s + t
|
|
return t.encode("utf-8")
|
|
|
|
|
|
@app.route("/")
|
|
def index():
|
|
q = request.args.get("q", "").strip()
|
|
t = get_tickers()
|
|
if q:
|
|
q2 = q.upper()
|
|
c = {k: v for k, v in t.items() if q2 in k.upper() or q2 in v["name"].upper()}
|
|
else:
|
|
c = dict(list(t.items())[:50])
|
|
q = ""
|
|
return render_template("index.html", companies=c, query=q, total=len(c))
|
|
|
|
|
|
@app.route("/<ticker>")
|
|
def company(ticker):
|
|
t = get_tickers()
|
|
if ticker not in t:
|
|
abort(404)
|
|
info = t[ticker]
|
|
ff = request.args.get("form", "").upper()
|
|
forms = {"10-K"} if ff == "10-K" else {"10-Q"} if ff == "10-Q" else {"10-K", "10-Q"}
|
|
try:
|
|
all_f = get_filings(info["cik"], forms)
|
|
except Exception as e:
|
|
log.error(f"Failed {ticker}: {e}")
|
|
abort(503, description=str(e))
|
|
p = request.args.get("page", 1, type=int)
|
|
s = (p - 1) * PAGE_SIZE
|
|
e = s + PAGE_SIZE
|
|
page = all_f[s:e]
|
|
tp = (len(all_f) + PAGE_SIZE - 1) // PAGE_SIZE
|
|
return render_template("company.html", ticker=ticker, company=info, cik=info["cik"],
|
|
filings=page, form_filter=ff, page=p, total_pages=tp, total=len(all_f))
|
|
|
|
|
|
@app.route("/<ticker>/view/<filename>")
|
|
def view(ticker, filename):
|
|
t = get_tickers()
|
|
if ticker not in t:
|
|
abort(404)
|
|
cik = t[ticker]["cik"]
|
|
dest = CACHE_DIR / "filings" / cik / filename
|
|
dest.parent.mkdir(parents=True, exist_ok=True)
|
|
if not dest.exists():
|
|
try:
|
|
fs = get_filings(cik, {"10-K", "10-Q"})
|
|
url = next((f["url"] for f in fs if f["file"] == filename), None)
|
|
if not url:
|
|
abort(404, description="Not found")
|
|
log.info(f"Caching {filename}...")
|
|
dest.write_bytes(fetch_filing(url))
|
|
except Exception as e:
|
|
log.error(f"Cache fail {ticker}/{filename}: {e}")
|
|
abort(503, description=str(e))
|
|
if request.args.get("reader", "").lower() in ("1", "true"):
|
|
return make_response(clean_ereader(dest.read_bytes())), {"Content-Type": "text/html"}
|
|
return send_from_directory(str(dest.parent), filename, mimetype="text/html")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
CACHE_DIR.mkdir(parents=True, exist_ok=True)
|
|
app.run(host="0.0.0.0", port=5001)
|