""" 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())