import json import os import re import urllib.error import urllib.parse import urllib.request from datetime import datetime DISPLAY_DATE_FORMAT = '%d-%m-%Y' K_CAMERA_PATTERN = re.compile(r'^K(\d+)\b', re.IGNORECASE) KANDANG_NUMBER_PATTERN = re.compile(r'kandang\s*(\d+)\s*$', re.IGNORECASE) def normalize_base_url(raw_url): url = str(raw_url or '').strip().rstrip('/') if not url: return 'http://localhost:5002' if not url.startswith(('http://', 'https://')): url = f'http://{url}' return url def get_cycle_api_config(): base_url = normalize_base_url(os.environ.get('CYCLE_API_BASE_URL', 'http://localhost:5002')) api_key = os.environ.get('CYCLE_API_KEY', '').strip() return base_url, api_key def is_cycle_api_configured(): _, api_key = get_cycle_api_config() return bool(api_key) def fetch_paginated_results(path, *, timeout=15): base_url, api_key = get_cycle_api_config() if not api_key: raise RuntimeError('CYCLE_API_KEY is not configured.') results = [] next_url = urllib.parse.urljoin(f'{base_url}/', path.lstrip('/')) while next_url: request = urllib.request.Request(next_url, method='GET') request.add_header('Accept', 'application/json') request.add_header('X-API-Key', api_key) try: with urllib.request.urlopen(request, timeout=timeout) as response: payload = json.loads(response.read().decode('utf-8')) except urllib.error.HTTPError as error: body = error.read().decode('utf-8', errors='replace') raise RuntimeError(f'Cycle API request failed ({error.code}): {body}') from error except (urllib.error.URLError, TimeoutError, json.JSONDecodeError) as error: raise RuntimeError(f'Cycle API request failed: {error}') from error if isinstance(payload, dict) and 'results' in payload: page_results = payload.get('results') or [] if not isinstance(page_results, list): raise RuntimeError('Cycle API paginated response is invalid.') results.extend(page_results) next_url = payload.get('next') continue if isinstance(payload, list): results.extend(payload) next_url = None continue raise RuntimeError('Cycle API response format is not supported.') return results def extract_k_index(source_name): match = K_CAMERA_PATTERN.match(str(source_name or '').strip()) if not match: return None return int(match.group(1)) def extract_kandang_number(kandang_name): match = KANDANG_NUMBER_PATTERN.search(str(kandang_name or '').strip()) if not match: return None return int(match.group(1)) def find_kandang_by_k_index(kandangs, k_index): if k_index is None: return None for kandang in kandangs: kandang_name = str(kandang.get('kandang_name', '')).strip() number_match = KANDANG_NUMBER_PATTERN.search(kandang_name) if number_match and int(number_match.group(1)) == k_index: return kandang return None def format_dashboard_date(iso_date): if not iso_date: return '' return datetime.strptime(iso_date, '%Y-%m-%d').strftime(DISPLAY_DATE_FORMAT) def select_cycle_for_kandang(cycles, kandang_id): kandang_cycles = [ cycle for cycle in cycles if cycle.get('kandang') == kandang_id ] if not kandang_cycles: return None active_cycles = [cycle for cycle in kandang_cycles if cycle.get('status') == 'active'] candidate_cycles = active_cycles or kandang_cycles return max(candidate_cycles, key=lambda cycle: cycle.get('id', 0)) def format_cycle_name(cycle_start, cycle_end): start = str(cycle_start or '').strip() end = str(cycle_end or '').strip() if start and end: return f'{start} - {end}' if start: return f'{start} - akhir data' if end: return f'awal data - {end}' return 'Siklus Aktif' def build_cycle_settings(cycle, kandang): kandang_name = kandang.get('kandang_name', f"Kandang {kandang.get('id')}") cycle_start = format_dashboard_date(cycle.get('start_date')) cycle_end = format_dashboard_date(cycle.get('end_date')) return { 'id': cycle.get('id', ''), 'kandang_id': kandang.get('id'), 'k_index': extract_kandang_number(kandang_name), 'kandang_name': kandang_name, 'cycle_name': format_cycle_name(cycle_start, cycle_end), 'cycle_start': cycle_start, 'cycle_end': cycle_end, 'saldo_awal': cycle.get('feed_initial_balance') or 0, 'feed_initial_balance_date': format_dashboard_date(cycle.get('feed_initial_balance_date')), 'status': cycle.get('status', ''), 'flock': cycle.get('flock'), } def build_all_cycles_by_kandang(kandangs, cycles): cycles_by_kandang = {} for kandang in kandangs: kandang_id = kandang.get('id') if kandang_id is None: continue kandang_cycles = [ cycle for cycle in cycles if cycle.get('kandang') == kandang_id ] kandang_cycles.sort(key=lambda cycle: cycle.get('id', 0), reverse=True) cycles_by_kandang[kandang_id] = [ build_cycle_settings(cycle, kandang) for cycle in kandang_cycles ] return cycles_by_kandang def load_kandang_cycle_context(): kandangs = fetch_paginated_results('/api/v1/kandangs/') cycles = fetch_paginated_results('/api/v1/cycles/') kandang_by_k_index = {} cycle_by_kandang_id = {} kandang_cycles = [] for kandang in kandangs: k_index = extract_kandang_number(kandang.get('kandang_name')) if k_index is not None: kandang_by_k_index[k_index] = kandang cycle = select_cycle_for_kandang(cycles, kandang.get('id')) if not cycle: continue cycle_settings = build_cycle_settings(cycle, kandang) cycle_by_kandang_id[kandang['id']] = cycle_settings kandang_cycles.append(cycle_settings) kandang_cycles.sort( key=lambda item: (item.get('k_index') is None, item.get('k_index') or 0, item.get('kandang_id') or 0) ) return { 'kandangs': kandangs, 'kandang_by_k_index': kandang_by_k_index, 'cycle_by_kandang_id': cycle_by_kandang_id, 'kandang_cycles': kandang_cycles, 'all_cycles_by_kandang': build_all_cycles_by_kandang(kandangs, cycles), } def get_cycle_settings_for_camera(source_name, cycle_context): k_index = extract_k_index(source_name) if k_index is None: return None kandang = find_kandang_by_k_index(cycle_context.get('kandangs', []), k_index) if not kandang: return None return cycle_context.get('cycle_by_kandang_id', {}).get(kandang['id'])