| 1603 | |
| 1604 | @shared_task(time_limit=3600, soft_time_limit=3500) |
| 1605 | def parse_programs_for_tvg_id(epg_id, force=False): |
| 1606 | try: |
| 1607 | from apps.epg.models import EPGData |
| 1608 | epg_obj = EPGData.objects.select_related('epg_source').filter(id=epg_id).first() |
| 1609 | if epg_obj and epg_obj.epg_source and epg_obj.epg_source.source_type == 'schedules_direct': |
| 1610 | return fetch_sd_guide_for_epg(epg_id, force=force) |
| 1611 | except Exception as e: |
| 1612 | logger.warning(f"Could not check EPG source type for id={epg_id}: {e}") |
| 1613 | |
| 1614 | if not acquire_task_lock('parse_epg_programs', epg_id): |
| 1615 | logger.info(f"Program parse for {epg_id} already in progress, skipping duplicate task") |
| 1616 | return "Task already running" |
| 1617 | |
| 1618 | lock_renewer = TaskLockRenewer('parse_epg_programs', epg_id) |
| 1619 | lock_renewer.start() |
| 1620 | |
| 1621 | source_file = None |
| 1622 | program_parser = None |
| 1623 | programs_to_create = [] |
| 1624 | programs_processed = 0 |
| 1625 | try: |
| 1626 | # Add memory tracking only in trace mode or higher |
| 1627 | try: |
| 1628 | process = None |
| 1629 | # Get current log level as a number |
| 1630 | current_log_level = logger.getEffectiveLevel() |
| 1631 | |
| 1632 | # Only track memory usage when log level is TRACE or more verbose or if running in DEBUG mode |
| 1633 | should_log_memory = current_log_level <= 5 or settings.DEBUG |
| 1634 | |
| 1635 | if should_log_memory: |
| 1636 | process = psutil.Process() |
| 1637 | initial_memory = process.memory_info().rss / 1024 / 1024 |
| 1638 | logger.info(f"[parse_programs_for_tvg_id] Initial memory usage: {initial_memory:.2f} MB") |
| 1639 | mem_before = initial_memory |
| 1640 | except ImportError: |
| 1641 | process = None |
| 1642 | should_log_memory = False |
| 1643 | |
| 1644 | epg = EPGData.objects.get(id=epg_id) |
| 1645 | epg_source = epg.epg_source |
| 1646 | |
| 1647 | # Skip program parsing for dummy EPG sources - they don't have program data files |
| 1648 | if epg_source.source_type == 'dummy': |
| 1649 | logger.info(f"Skipping program parsing for dummy EPG source {epg_source.name} (ID: {epg_id})") |
| 1650 | lock_renewer.stop() |
| 1651 | release_task_lock('parse_epg_programs', epg_id) |
| 1652 | return |
| 1653 | |
| 1654 | if not force and not Channel.objects.filter(epg_data=epg).exists(): |
| 1655 | logger.info(f"No channels matched to EPG {epg.tvg_id}") |
| 1656 | lock_renewer.stop() |
| 1657 | release_task_lock('parse_epg_programs', epg_id) |
| 1658 | return |
| 1659 | |
| 1660 | logger.info(f"Refreshing program data for tvg_id: {epg.tvg_id}") |
| 1661 | |
| 1662 | # Optimize deletion with a single delete query instead of chunking |