jessikitty
f7c20de5ac
LIM configured no logging at all, so the instruction fetcher had no trail to follow. Adds a rotating file handler on the root logger plus two owner-only admin pages. - Admin > Logs: filter by level, source and text; tracebacks kept with their error; auto-refresh, raw download, rotated-file browsing and rotate-now - Admin > Logs > Brickset API calls: every outbound call recorded with method, params, caller, duration, result count and error, plus daily quota accounting (only getSets counts) and a comparison against getKeyUsageStats - New owner concept (services/ownership.py): a single account, distinct from is_admin, since these pages expose other admins activity. Defaults to the first admin; flask owner show/set to change - Services now log through module loggers; retention setting and pruning for call records - New CLI: flask owner show/set, flask logs path/prune - New table brickset_api_calls, created automatically at boot; LOG_* env vars, /data/logs in Docker
229 lines
9.3 KiB
Python
229 lines
9.3 KiB
Python
"""
|
|
Instruction fetcher diagnostics - read-only.
|
|
|
|
Drop this in your LIM folder (next to run.py) and run it. It changes nothing;
|
|
it just reports what the fetcher can see, in the order things have to be true
|
|
for a download to happen.
|
|
|
|
E:\\LIM> venv\\Scripts\\activate
|
|
E:\\LIM> python diagnose_fetcher.py
|
|
|
|
Optionally test a live Brickset lookup for one set (no files are saved):
|
|
|
|
E:\\LIM> python diagnose_fetcher.py 10497
|
|
|
|
Docker:
|
|
|
|
docker compose exec app python diagnose_fetcher.py
|
|
docker compose exec app python diagnose_fetcher.py 10497
|
|
|
|
Created by: Jess Rogerson (yelling commands at Claude.AI)
|
|
"""
|
|
|
|
import os
|
|
import sys
|
|
|
|
os.environ.setdefault('INSTRUCTION_FETCH_WORKER', '0')
|
|
|
|
|
|
def line(char='-'):
|
|
print(char * 68)
|
|
|
|
|
|
def head(text):
|
|
print()
|
|
line('=')
|
|
print(f' {text}')
|
|
line('=')
|
|
|
|
|
|
def main():
|
|
test_set = sys.argv[1] if len(sys.argv) > 1 else None
|
|
|
|
try:
|
|
from app import create_app, db
|
|
except Exception as exc:
|
|
print(f'FAILED to import the app: {exc}')
|
|
print('Are you in the LIM folder with the virtualenv activated?')
|
|
return 1
|
|
|
|
app = create_app(os.environ.get('FLASK_ENV', 'production'))
|
|
|
|
with app.app_context():
|
|
# -- is the update even installed? ---------------------------------
|
|
head('1. Is the fetcher installed?')
|
|
try:
|
|
from app.models.instruction_fetch import InstructionFetchJob
|
|
from app.services import instruction_fetcher as fetcher
|
|
print(' code : present')
|
|
except Exception as exc:
|
|
print(f' code : MISSING ({exc})')
|
|
print('\n The update is not installed in this folder.')
|
|
return 1
|
|
|
|
from sqlalchemy import inspect
|
|
inspector = inspect(db.engine)
|
|
tables = inspector.get_table_names()
|
|
print(f" jobs table : "
|
|
f"{'present' if 'instruction_fetch_jobs' in tables else 'MISSING - restart LIM'}")
|
|
cols = [c['name'] for c in inspector.get_columns('instructions')]
|
|
print(f" instructions.sha256: {'present' if 'sha256' in cols else 'MISSING - restart LIM'}")
|
|
|
|
routes = [r.rule for r in app.url_map.iter_rules()
|
|
if 'instruction-fetch' in r.rule]
|
|
print(f' admin routes : {len(routes)} registered')
|
|
|
|
# -- credentials ----------------------------------------------------
|
|
head('2. Brickset credentials')
|
|
from app.services.brickset_api import BricksetAPI
|
|
configured = BricksetAPI.is_configured()
|
|
for key in ('BRICKSET_API_KEY', 'BRICKSET_USERNAME', 'BRICKSET_PASSWORD'):
|
|
value = app.config.get(key)
|
|
shown = ('set (%d chars)' % len(value)) if value else 'NOT SET'
|
|
print(f' {key:<18} : {shown}')
|
|
print(f' is_configured() : {configured}')
|
|
|
|
# -- settings --------------------------------------------------------
|
|
head('3. Fetcher settings')
|
|
from app.models.site_setting import SiteSetting
|
|
for key in sorted(fetcher.SETTING_DEFAULTS):
|
|
print(f' {key:<26} : {SiteSetting.get(key, "(unset)")!r}')
|
|
print()
|
|
print(f' is_enabled() : {fetcher.is_enabled()}')
|
|
print(f' interval_minutes() : {fetcher.interval_minutes()}')
|
|
print(f' is_due() : {fetcher.is_due()}')
|
|
print(f' seconds_until_due() : {fetcher.seconds_until_due()}')
|
|
print(f' last_run() : {fetcher.last_run()}')
|
|
|
|
# -- the library ------------------------------------------------------
|
|
head('4. What the queue builder can see')
|
|
from app.models.set import Set
|
|
from app.models.instruction import Instruction
|
|
|
|
total = Set.query.count()
|
|
official = fetcher.official_sets_query().count()
|
|
mocs = Set.query.filter(Set.is_moc.is_(True)).count()
|
|
null_moc = Set.query.filter(Set.is_moc.is_(None)).count()
|
|
missing = fetcher.sets_missing_pdfs().count()
|
|
|
|
print(f' sets total : {total}')
|
|
print(f' official (is_moc = false) : {official}')
|
|
print(f' MOCs (is_moc = true) : {mocs}')
|
|
print(f' is_moc IS NULL : {null_moc} '
|
|
f'{"<-- these are invisible to the fetcher!" if null_moc else ""}')
|
|
print(f' official with no PDF : {missing} <-- "Queue sets missing instructions" queues this many')
|
|
|
|
print()
|
|
print(' file_type values in the instructions table:')
|
|
rows = db.session.query(
|
|
Instruction.file_type, db.func.count(Instruction.id)
|
|
).group_by(Instruction.file_type).all()
|
|
if rows:
|
|
for file_type, count in rows:
|
|
flag = '' if file_type == 'PDF' else ''
|
|
print(f' {file_type!r:<12} {count}{flag}')
|
|
else:
|
|
print(' (no instruction files at all)')
|
|
|
|
if official:
|
|
print()
|
|
print(' first few official sets:')
|
|
for lego_set in fetcher.official_sets_query().limit(5):
|
|
pdfs = lego_set.instructions.filter_by(file_type='PDF').count()
|
|
print(f' {lego_set.set_number:<12} {(lego_set.set_name or "")[:32]:<34} '
|
|
f'PDFs={pdfs} brickset_id={lego_set.brickset_id}')
|
|
|
|
# -- the queue ---------------------------------------------------------
|
|
head('5. The queue')
|
|
counts = InstructionFetchJob.counts()
|
|
for key, value in counts.items():
|
|
print(f' {key:<12} : {value}')
|
|
|
|
recent = InstructionFetchJob.query.order_by(
|
|
InstructionFetchJob.id.desc()).limit(10).all()
|
|
if recent:
|
|
print()
|
|
print(' most recent jobs:')
|
|
for job in recent:
|
|
number = job.lego_set.set_number if job.lego_set else '(deleted)'
|
|
print(f' #{job.id:<5} {number:<12} {job.status:<10} '
|
|
f'files={job.files_added} tries={job.attempts}')
|
|
if job.message:
|
|
print(f' {job.message[:100]}')
|
|
else:
|
|
print()
|
|
print(' No jobs have ever been queued.')
|
|
|
|
# -- verdict -------------------------------------------------------------
|
|
head('6. Verdict')
|
|
problems = []
|
|
if not configured:
|
|
problems.append('Brickset credentials are not set in .env')
|
|
if null_moc:
|
|
problems.append(f'{null_moc} set(s) have is_moc = NULL and are skipped')
|
|
if not official:
|
|
problems.append('there are no official sets in the library')
|
|
elif not missing and counts['total'] == 0:
|
|
problems.append('every official set already has a PDF, so '
|
|
'"queue missing" finds nothing - use '
|
|
'"Queue all official sets" instead')
|
|
if not fetcher.is_enabled():
|
|
problems.append('the fetcher is switched OFF (admin page, '
|
|
'"Work through the queue automatically")')
|
|
if counts['queued'] and not fetcher.is_enabled():
|
|
problems.append(f"{counts['queued']} job(s) are waiting but "
|
|
f"nothing will run until it is switched on")
|
|
if not app.config.get('INSTRUCTION_FETCH_WORKER', True):
|
|
problems.append('INSTRUCTION_FETCH_WORKER is 0, so the web '
|
|
'process runs no worker thread')
|
|
|
|
if problems:
|
|
for problem in problems:
|
|
print(f' * {problem}')
|
|
else:
|
|
print(' Nothing obviously wrong. If the queue has jobs and the '
|
|
'fetcher is on,')
|
|
print(' the next one runs in '
|
|
f'{fetcher.seconds_until_due()} seconds.')
|
|
|
|
# -- live test ------------------------------------------------------------
|
|
if test_set:
|
|
head(f'7. Live Brickset lookup for {test_set}')
|
|
number = fetcher.brickset_number(test_set)
|
|
print(f' asking Brickset for : {number}')
|
|
api = BricksetAPI()
|
|
try:
|
|
entries = api.get_instructions(number)
|
|
print(f' getInstructions2 : {len(entries)} result(s)')
|
|
for entry in entries:
|
|
print(f' {entry.get("description", "")}')
|
|
print(f' {entry.get("URL", "")}')
|
|
if not entries:
|
|
print(' Nothing returned. Either Brickset has no '
|
|
'instructions for this set,')
|
|
print(' or the credentials/number are wrong.')
|
|
else:
|
|
import requests
|
|
url = entries[0].get('URL')
|
|
print()
|
|
print(' testing the first download (headers only)...')
|
|
response = requests.get(
|
|
url, stream=True, timeout=30,
|
|
headers={'User-Agent': fetcher.USER_AGENT})
|
|
print(f' HTTP {response.status_code} '
|
|
f'{response.headers.get("Content-Type")} '
|
|
f'{response.headers.get("Content-Length")} bytes')
|
|
response.close()
|
|
except Exception as exc:
|
|
print(f' FAILED: {type(exc).__name__}: {exc}')
|
|
|
|
print()
|
|
line('=')
|
|
print(' end of report')
|
|
line('=')
|
|
return 0
|
|
|
|
|
|
if __name__ == '__main__':
|
|
sys.exit(main())
|