Merge pull request 'Dev' (#12) from Dev into main

Reviewed-on: #12

README.md

@@ -8,9 +8,11 @@ Servicio Python que genera y actualiza automáticamente listas de reproducción
 
 ## Características
 - **Actualización automática** — Regenera la playlist en intervalos configurables
-- **Servidor HTTP integrado** — Sirve el archivo M3U directamente sin dependencias externas
+- **EPG integrado** — Descarga y filtra automáticamente la guía de programación (XMLTV)
+- **Filtros avanzados** — Incluye canales por prefijo o excluye por contenido del nombre
+- **Servidor HTTP integrado** — Sirve el archivo M3U y EPG directamente sin dependencias externas
 - **Docker ready** — Despliegue simple con Docker Compose
-- **Escritura atómica** — Actualiza el archivo sin interrumpir descargas activas
+- **Escritura atómica** — Actualiza archivos sin interrumpir descargas activas
 - **Seguridad** — Se ejecuta con usuario no privilegiado en contenedor
 - **Multi-hilo** — Servidor threaded para manejar múltiples clientes simultáneos
 
@@ -41,9 +43,10 @@ Servicio Python que genera y actualiza automáticamente listas de reproducción
    docker compose up -d
    ```
 
-4. **Accede a la playlist:**
+4. **Accede a la playlist y EPG:**
    ```
    http://localhost:8080/playlist.m3u
+   http://localhost:8080/epg.xml
    ```
 
 ### Sin Docker
@@ -74,6 +77,21 @@ Servicio Python que genera y actualiza automáticamente listas de reproducción
 | `PORT` | Puerto del servidor HTTP local | ❌ | `8080` |
 | `UPDATE_INTERVAL` | Intervalo de actualización (segundos) | ❌ | `3600` |
 | `OUTPUT_FILE` | Nombre del archivo M3U generado | ❌ | `playlist.m3u` |
+| `INCLUDE_TEXT` | Lista JSON de prefijos. Solo canales que **empiecen** con alguno | ❌ | `[]` |
+| `EXCLUDE_TEXT` | Lista JSON de textos. Excluye canales que **contengan** alguno | ❌ | `[]` |
+
+### Filtros
+
+- **`INCLUDE_TEXT`**: Filtra canales que **empiecen** con alguno de los textos especificados (lógica OR). Si está vacío, incluye todos los canales.
+- **`EXCLUDE_TEXT`**: Excluye canales cuyo nombre **contenga** alguno de los textos (lógica OR). Se aplica después del filtro de inclusión.
+- Los mismos filtros se aplican automáticamente al EPG.
+
+Ejemplo:
+```env
+# Solo canales que empiecen con "ES:" o "UK:", excluyendo los que contengan "Adult"
+INCLUDE_TEXT=["ES:", "UK:"]
+EXCLUDE_TEXT=["Adult", "XXX"]
+```
 
 ## Arquitectura
 
@@ -87,10 +105,15 @@ Servicio Python que genera y actualiza automáticamente listas de reproducción
 │  │ Fetch API      │  │  │ ThreadingHTTPServer        │  │
 │  │ (Xtream Codes) │  │  │                            │  │
 │  └───────┬────────┘  │  │  GET /playlist.m3u         │  │
-│          │           │  │                            │  │
+│          │           │  │  GET /epg.xml              │  │
 │  ┌───────▼────────┐  │  └────────────────────────────┘  │
 │  │ Generate M3U   │  │                                  │
-│  │ (Atomic Write) │──┼──────► public/playlist.m3u      │
+│  │ + Apply Filter │──┼──────► public/playlist.m3u      │
+│  └───────┬────────┘  │                                  │
+│          │           │                                  │
+│  ┌───────▼────────┐  │                                  │
+│  │ Fetch & Filter │  │                                  │
+│  │ EPG (XMLTV)    │──┼──────► public/epg.xml           │
 │  └───────┬────────┘  │                                  │
 │          │           │                                  │
 │     sleep(interval)  │                                  │

pyproject.toml

@@ -1,6 +1,6 @@
 [tool.poetry]
 name = "m3u_list_builder"
-version = "0.0.0"
+version = "0.3.2"
 description = "Python tool to build M3U lists from various sources."
 authors = ["Unai Blazquez <unaibg2000@gmail.com>"]
 readme = "README.md"

src/m3u_list_builder/playlist.py

@@ -2,6 +2,7 @@
 
 import logging
 import time
+import xml.etree.ElementTree as ET
 from pathlib import Path
 
 import requests
@@ -24,7 +25,7 @@ class PlaylistManager:
         PUBLIC_DIR.mkdir(exist_ok=True)
 
     def fetch_and_generate(self):
-        """Descarga datos y regenera el archivo M3U."""
+        """Descarga datos y regenera el archivo M3U y EPG."""
         logger.info("Iniciando actualización de playlist...")
 
         url = f"{settings.host}/player_api.php"
@@ -44,11 +45,95 @@ class PlaylistManager:
                 f"Playlist actualizada exitosamente. Total canales: {len(data)}"
             )
 
+            # Fetch and write EPG data
+            self.fetch_and_write_epg()
+
         except requests.RequestException as e:
             logger.error(f"Error de red al obtener playlist: {e}")
         except Exception as e:
             logger.exception(f"Error inesperado actualizando playlist: {e}")
 
+    def fetch_and_write_epg(self):
+        """Descarga y guarda los datos EPG desde el servidor XMLTV."""
+        logger.info("Iniciando descarga de EPG...")
+
+        url = f"{settings.host}/xmltv.php"
+        params = {
+            "username": settings.username,
+            "password": settings.password,
+        }
+
+        try:
+            response = requests.get(url, params=params, timeout=60)
+            response.raise_for_status()
+
+            # Parse and filter EPG XML
+            filtered_epg = self._filter_epg(response.content)
+            self._write_epg(filtered_epg)
+
+            logger.info("EPG actualizado exitosamente.")
+
+        except requests.RequestException as e:
+            logger.error(f"Error de red al obtener EPG: {e}")
+        except ET.ParseError as e:
+            logger.error(f"Error parseando XML de EPG: {e}")
+        except Exception as e:
+            logger.exception(f"Error inesperado actualizando EPG: {e}")
+
+    def _filter_epg(self, xml_content: bytes) -> ET.Element:
+        """Filtra el EPG para incluir solo canales que pasan los filtros."""
+        root = ET.fromstring(xml_content)
+
+        # Get list of channel IDs to keep based on filters
+        channels_to_remove = []
+        for channel in root.findall("channel"):
+            display_name = channel.find("display-name")
+            name = display_name.text if display_name is not None else ""
+
+            should_remove = False
+
+            # Apply include filter (startswith)
+            if settings.include_text:
+                if not any(
+                    name.lower().startswith(inc_text.lower())
+                    for inc_text in settings.include_text
+                ):
+                    should_remove = True
+
+            # Apply exclude filter (contains)
+            if not should_remove and settings.exclude_text:
+                if any(
+                    exc_text.lower() in name.lower()
+                    for exc_text in settings.exclude_text
+                ):
+                    should_remove = True
+
+            if should_remove:
+                channels_to_remove.append(channel.get("id"))
+
+        # Remove filtered channels
+        for channel in root.findall("channel"):
+            if channel.get("id") in channels_to_remove:
+                root.remove(channel)
+
+        # Remove programmes for filtered channels
+        for programme in root.findall("programme"):
+            if programme.get("channel") in channels_to_remove:
+                root.remove(programme)
+
+        return root
+
+    def _write_epg(self, epg_root: ET.Element):
+        """Escribe el archivo EPG en disco de forma atómica."""
+        temp_file = PUBLIC_DIR / "epg.xml.tmp"
+        final_file = PUBLIC_DIR / "epg.xml"
+
+        tree = ET.ElementTree(epg_root)
+        tree.write(temp_file, encoding="utf-8", xml_declaration=True)
+
+        # Reemplazo atómico
+        temp_file.replace(final_file)
+
     def _write_m3u(self, channels: list):
         """Escribe el archivo M3U en disco de forma atómica."""
         temp_file = PUBLIC_DIR / f"{settings.output_file}.tmp"
@@ -61,6 +146,9 @@ class PlaylistManager:
                 stream_id = channel.get("stream_id")
                 icon = channel.get("stream_icon", "")
                 cat_id = channel.get("category_id", "")
+                epg_id = channel.get("epg_channel_id", "")
+                if not epg_id:
+                    epg_id = name
                 # Filtros de inclusión/exclusión por prefijo de nombre
                 if settings.include_text:
                     if not any(
@@ -81,7 +169,7 @@ class PlaylistManager:
                 )
 
                 extinf_line = (
-                    f'#EXTINF:-1 tvg-id="{name}" tvg-logo="{icon}" '
+                    f'#EXTINF:-1 tvg-id="{epg_id}" tvg-logo="{icon}" '
                     f'group-title="Cat_{cat_id}",{name}\n'
                 )
                 f.write(extinf_line)

tests/test_playlist.py

@@ -66,10 +66,11 @@ class TestPlaylistManager:
                 mock_get.return_value = mock_response
 
                 with patch.object(manager, "_write_m3u") as mock_write:
-                    manager.fetch_and_generate()
+                    with patch.object(manager, "fetch_and_write_epg"):
+                        manager.fetch_and_generate()
 
-                    mock_get.assert_called_once()
-                    mock_write.assert_called_once_with(sample_channels)
+                        mock_get.assert_called_once()
+                        mock_write.assert_called_once_with(sample_channels)
 
     def test_fetch_and_generate_correct_api_call(self):
         """Test: fetch_and_generate llama a la API correctamente."""
@@ -89,17 +90,18 @@ class TestPlaylistManager:
                 mock_get.return_value = mock_response
 
                 with patch.object(manager, "_write_m3u"):
-                    manager.fetch_and_generate()
+                    with patch.object(manager, "fetch_and_write_epg"):
+                        manager.fetch_and_generate()
 
-                    mock_get.assert_called_once_with(
-                        "http://test-iptv.com/player_api.php",
-                        params={
-                            "username": "testuser",
-                            "password": "testpass",
-                            "action": "get_live_streams",
-                        },
-                        timeout=30,
-                    )
+                        mock_get.assert_called_once_with(
+                            "http://test-iptv.com/player_api.php",
+                            params={
+                                "username": "testuser",
+                                "password": "testpass",
+                                "action": "get_live_streams",
+                            },
+                            timeout=30,
+                        )
 
     def test_fetch_and_generate_request_exception(self, caplog):
         """Test: fetch_and_generate maneja RequestException sin crash."""
@@ -302,6 +304,104 @@ class TestPlaylistManager:
                 # Verificar URL
                 assert lines[2] == "http://iptv.com/live/user/pass/123.ts"
 
+    def test_write_m3u_uses_epg_channel_id_when_present(self, temp_dir):
+        """Test: _write_m3u usa epg_channel_id en tvg-id cuando está presente."""
+        channel = [
+            {
+                "name": "Test Channel",
+                "stream_id": 123,
+                "stream_icon": "http://icon.com/test.png",
+                "category_id": "5",
+                "epg_channel_id": "test.channel.epg",
+            }
+        ]
+
+        with patch("m3u_list_builder.playlist.PUBLIC_DIR", temp_dir):
+            with patch("m3u_list_builder.playlist.settings") as mock_settings:
+                mock_settings.host = "http://iptv.com"
+                mock_settings.username = "user"
+                mock_settings.password = "pass"
+                mock_settings.output_file = "playlist.m3u"
+                mock_settings.include_text = []
+                mock_settings.exclude_text = []
+
+                from m3u_list_builder.playlist import PlaylistManager
+
+                manager = PlaylistManager()
+                manager._write_m3u(channel)
+
+                output_file = temp_dir / "playlist.m3u"
+                content = output_file.read_text()
+
+                # Debe usar epg_channel_id en tvg-id
+                assert 'tvg-id="test.channel.epg"' in content
+                # El nombre debe seguir apareciendo al final de EXTINF
+                assert ",Test Channel" in content
+
+    def test_write_m3u_falls_back_to_name_when_epg_channel_id_empty(self, temp_dir):
+        """Test: _write_m3u usa name como tvg-id cuando epg_channel_id está vacío."""
+        channel = [
+            {
+                "name": "Fallback Channel",
+                "stream_id": 456,
+                "stream_icon": "",
+                "category_id": "1",
+                "epg_channel_id": "",
+            }
+        ]
+
+        with patch("m3u_list_builder.playlist.PUBLIC_DIR", temp_dir):
+            with patch("m3u_list_builder.playlist.settings") as mock_settings:
+                mock_settings.host = "http://iptv.com"
+                mock_settings.username = "user"
+                mock_settings.password = "pass"
+                mock_settings.output_file = "playlist.m3u"
+                mock_settings.include_text = []
+                mock_settings.exclude_text = []
+
+                from m3u_list_builder.playlist import PlaylistManager
+
+                manager = PlaylistManager()
+                manager._write_m3u(channel)
+
+                output_file = temp_dir / "playlist.m3u"
+                content = output_file.read_text()
+
+                # Debe usar name como fallback en tvg-id
+                assert 'tvg-id="Fallback Channel"' in content
+
+    def test_write_m3u_falls_back_to_name_when_epg_channel_id_missing(self, temp_dir):
+        """Test: _write_m3u usa name como tvg-id cuando epg_channel_id no existe."""
+        channel = [
+            {
+                "name": "No EPG Channel",
+                "stream_id": 789,
+                "stream_icon": "",
+                "category_id": "2",
+                # No epg_channel_id field at all
+            }
+        ]
+
+        with patch("m3u_list_builder.playlist.PUBLIC_DIR", temp_dir):
+            with patch("m3u_list_builder.playlist.settings") as mock_settings:
+                mock_settings.host = "http://iptv.com"
+                mock_settings.username = "user"
+                mock_settings.password = "pass"
+                mock_settings.output_file = "playlist.m3u"
+                mock_settings.include_text = []
+                mock_settings.exclude_text = []
+
+                from m3u_list_builder.playlist import PlaylistManager
+
+                manager = PlaylistManager()
+                manager._write_m3u(channel)
+
+                output_file = temp_dir / "playlist.m3u"
+                content = output_file.read_text()
+
+                # Debe usar name como fallback en tvg-id
+                assert 'tvg-id="No EPG Channel"' in content
+
     def test_write_m3u_missing_fields_uses_defaults(self, temp_dir, minimal_channel):
         """Test: _write_m3u usa valores por defecto para campos faltantes."""
         with patch("m3u_list_builder.playlist.PUBLIC_DIR", temp_dir):
@@ -830,3 +930,419 @@ class TestPlaylistManager:
                 # "News ABC" y "ZABC Channel" NO deben estar
                 assert "News ABC" not in content
                 assert "ZABC Channel" not in content
+
+    # ========================================
+    # Tests para fetch_and_write_epg - EPG Functionality
+    # ========================================
+
+    def test_fetch_and_write_epg_success(self, temp_dir):
+        """Test: fetch_and_write_epg descarga y guarda EPG correctamente."""
+        sample_epg = b'''<?xml version="1.0" encoding="UTF-8"?>
+        <tv>
+            <channel id="ch1">
+                <display-name>ESPN Sports</display-name>
+            </channel>
+            <programme channel="ch1" start="20260203" stop="20260203">
+                <title>Sports Event</title>
+            </programme>
+        </tv>'''
+
+        with patch("m3u_list_builder.playlist.PUBLIC_DIR", temp_dir):
+            with patch("m3u_list_builder.playlist.settings") as mock_settings:
+                mock_settings.host = "http://test-iptv.com"
+                mock_settings.username = "testuser"
+                mock_settings.password = "testpass"
+                mock_settings.include_text = []
+                mock_settings.exclude_text = []
+
+                from m3u_list_builder.playlist import PlaylistManager
+
+                manager = PlaylistManager()
+
+                with patch("m3u_list_builder.playlist.requests.get") as mock_get:
+                    mock_response = MagicMock()
+                    mock_response.content = sample_epg
+                    mock_response.raise_for_status = MagicMock()
+                    mock_get.return_value = mock_response
+
+                    manager.fetch_and_write_epg()
+
+                    # Verify EPG file was written
+                    epg_file = temp_dir / "epg.xml"
+                    assert epg_file.exists()
+                    content = epg_file.read_text()
+                    assert "ESPN Sports" in content
+
+    def test_fetch_and_write_epg_correct_api_call(self, temp_dir):
+        """Test: fetch_and_write_epg llama a la API correctamente."""
+        sample_epg = b'<?xml version="1.0"?><tv></tv>'
+
+        with patch("m3u_list_builder.playlist.PUBLIC_DIR", temp_dir):
+            with patch("m3u_list_builder.playlist.settings") as mock_settings:
+                mock_settings.host = "http://test-iptv.com"
+                mock_settings.username = "testuser"
+                mock_settings.password = "testpass"
+                mock_settings.include_text = []
+                mock_settings.exclude_text = []
+
+                from m3u_list_builder.playlist import PlaylistManager
+
+                manager = PlaylistManager()
+
+                with patch("m3u_list_builder.playlist.requests.get") as mock_get:
+                    mock_response = MagicMock()
+                    mock_response.content = sample_epg
+                    mock_response.raise_for_status = MagicMock()
+                    mock_get.return_value = mock_response
+
+                    manager.fetch_and_write_epg()
+
+                    mock_get.assert_called_once_with(
+                        "http://test-iptv.com/xmltv.php",
+                        params={
+                            "username": "testuser",
+                            "password": "testpass",
+                        },
+                        timeout=60,
+                    )
+
+    def test_fetch_and_write_epg_request_exception(self, caplog):
+        """Test: fetch_and_write_epg maneja RequestException sin crash."""
+        with patch("m3u_list_builder.playlist.settings") as mock_settings:
+            mock_settings.host = "http://test-iptv.com"
+            mock_settings.username = "testuser"
+            mock_settings.password = "testpass"
+
+            from m3u_list_builder.playlist import PlaylistManager
+
+            manager = PlaylistManager()
+
+            with patch("m3u_list_builder.playlist.requests.get") as mock_get:
+                mock_get.side_effect = requests.RequestException("Connection failed")
+
+                manager.fetch_and_write_epg()
+
+                assert "Error de red al obtener EPG" in caplog.text
+
+    def test_fetch_and_write_epg_parse_error(self, caplog):
+        """Test: fetch_and_write_epg maneja error de parsing XML."""
+        with patch("m3u_list_builder.playlist.settings") as mock_settings:
+            mock_settings.host = "http://test-iptv.com"
+            mock_settings.username = "testuser"
+            mock_settings.password = "testpass"
+
+            from m3u_list_builder.playlist import PlaylistManager
+
+            manager = PlaylistManager()
+
+            with patch("m3u_list_builder.playlist.requests.get") as mock_get:
+                mock_response = MagicMock()
+                mock_response.content = b"invalid xml <not closed"
+                mock_response.raise_for_status = MagicMock()
+                mock_get.return_value = mock_response
+
+                manager.fetch_and_write_epg()
+
+                assert "Error parseando XML de EPG" in caplog.text
+
+    def test_fetch_and_write_epg_unexpected_exception(self, caplog):
+        """Test: fetch_and_write_epg maneja excepciones inesperadas."""
+        with patch("m3u_list_builder.playlist.settings") as mock_settings:
+            mock_settings.host = "http://test-iptv.com"
+            mock_settings.username = "testuser"
+            mock_settings.password = "testpass"
+            mock_settings.include_text = []
+            mock_settings.exclude_text = []
+
+            from m3u_list_builder.playlist import PlaylistManager
+
+            manager = PlaylistManager()
+
+            with patch("m3u_list_builder.playlist.requests.get") as mock_get:
+                mock_response = MagicMock()
+                mock_response.content = b'<?xml version="1.0"?><tv></tv>'
+                mock_response.raise_for_status = MagicMock()
+                mock_get.return_value = mock_response
+
+                with patch.object(
+                    manager, "_write_epg", side_effect=RuntimeError("Disk full")
+                ):
+                    manager.fetch_and_write_epg()
+
+                    assert "Error inesperado actualizando EPG" in caplog.text
+
+    # ========================================
+    # Tests para _filter_epg - Filtrado de EPG
+    # ========================================
+
+    def test_filter_epg_include_text_filters_by_prefix(self):
+        """Test: _filter_epg filtra canales que EMPIEZAN con el texto."""
+        sample_epg = b'''<?xml version="1.0" encoding="UTF-8"?>
+        <tv>
+            <channel id="ch1">
+                <display-name>ESPN Sports</display-name>
+            </channel>
+            <channel id="ch2">
+                <display-name>Sports Center ESPN</display-name>
+            </channel>
+            <channel id="ch3">
+                <display-name>HBO Movies</display-name>
+            </channel>
+            <programme channel="ch1" start="20260203" stop="20260203">
+                <title>Sports Event</title>
+            </programme>
+            <programme channel="ch2" start="20260203" stop="20260203">
+                <title>Center Show</title>
+            </programme>
+            <programme channel="ch3" start="20260203" stop="20260203">
+                <title>Movie</title>
+            </programme>
+        </tv>'''
+
+        with patch("m3u_list_builder.playlist.settings") as mock_settings:
+            mock_settings.host = "http://test-iptv.com"
+            mock_settings.username = "testuser"
+            mock_settings.password = "testpass"
+            mock_settings.include_text = ["ESPN"]
+            mock_settings.exclude_text = []
+
+            from m3u_list_builder.playlist import PlaylistManager
+            import xml.etree.ElementTree as ET
+
+            manager = PlaylistManager()
+            result = manager._filter_epg(sample_epg)
+
+            # Convert to string for easier assertion
+            result_str = ET.tostring(result, encoding="unicode")
+
+            # ESPN Sports starts with ESPN - should be included
+            assert "ESPN Sports" in result_str
+            # Sports Center ESPN does NOT start with ESPN - should be filtered
+            assert "Sports Center ESPN" not in result_str
+            # HBO Movies - should be filtered
+            assert "HBO Movies" not in result_str
+            # Programme for ch1 should remain
+            assert "Sports Event" in result_str
+            # Programmes for ch2 and ch3 should be removed
+            assert "Center Show" not in result_str
+            assert "Movie" not in result_str
+
+    def test_filter_epg_exclude_text_filters_by_contains(self):
+        """Test: _filter_epg excluye canales que CONTIENEN el texto."""
+        sample_epg = b'''<?xml version="1.0" encoding="UTF-8"?>
+        <tv>
+            <channel id="ch1">
+                <display-name>ESPN Sports</display-name>
+            </channel>
+            <channel id="ch2">
+                <display-name>Adult Content</display-name>
+            </channel>
+            <programme channel="ch1" start="20260203" stop="20260203">
+                <title>Sports Event</title>
+            </programme>
+            <programme channel="ch2" start="20260203" stop="20260203">
+                <title>Adult Show</title>
+            </programme>
+        </tv>'''
+
+        with patch("m3u_list_builder.playlist.settings") as mock_settings:
+            mock_settings.host = "http://test-iptv.com"
+            mock_settings.username = "testuser"
+            mock_settings.password = "testpass"
+            mock_settings.include_text = []
+            mock_settings.exclude_text = ["Adult"]
+
+            from m3u_list_builder.playlist import PlaylistManager
+            import xml.etree.ElementTree as ET
+
+            manager = PlaylistManager()
+            result = manager._filter_epg(sample_epg)
+
+            result_str = ET.tostring(result, encoding="unicode")
+
+            # ESPN Sports - should remain
+            assert "ESPN Sports" in result_str
+            # Adult Content - should be filtered
+            assert "Adult Content" not in result_str
+            # Programme for ch1 should remain
+            assert "Sports Event" in result_str
+            # Programme for ch2 should be removed
+            assert "Adult Show" not in result_str
+
+    def test_filter_epg_no_filters_keeps_all(self):
+        """Test: _filter_epg sin filtros mantiene todos los canales."""
+        sample_epg = b'''<?xml version="1.0" encoding="UTF-8"?>
+        <tv>
+            <channel id="ch1">
+                <display-name>Channel 1</display-name>
+            </channel>
+            <channel id="ch2">
+                <display-name>Channel 2</display-name>
+            </channel>
+        </tv>'''
+
+        with patch("m3u_list_builder.playlist.settings") as mock_settings:
+            mock_settings.host = "http://test-iptv.com"
+            mock_settings.username = "testuser"
+            mock_settings.password = "testpass"
+            mock_settings.include_text = []
+            mock_settings.exclude_text = []
+
+            from m3u_list_builder.playlist import PlaylistManager
+            import xml.etree.ElementTree as ET
+
+            manager = PlaylistManager()
+            result = manager._filter_epg(sample_epg)
+
+            result_str = ET.tostring(result, encoding="unicode")
+
+            assert "Channel 1" in result_str
+            assert "Channel 2" in result_str
+
+    def test_filter_epg_include_and_exclude_combined(self):
+        """Test: _filter_epg aplica include y exclude juntos."""
+        sample_epg = b'''<?xml version="1.0" encoding="UTF-8"?>
+        <tv>
+            <channel id="ch1">
+                <display-name>ESPN Sports</display-name>
+            </channel>
+            <channel id="ch2">
+                <display-name>ESPN Adult</display-name>
+            </channel>
+            <channel id="ch3">
+                <display-name>HBO Movies</display-name>
+            </channel>
+        </tv>'''
+
+        with patch("m3u_list_builder.playlist.settings") as mock_settings:
+            mock_settings.host = "http://test-iptv.com"
+            mock_settings.username = "testuser"
+            mock_settings.password = "testpass"
+            mock_settings.include_text = ["ESPN"]
+            mock_settings.exclude_text = ["Adult"]
+
+            from m3u_list_builder.playlist import PlaylistManager
+            import xml.etree.ElementTree as ET
+
+            manager = PlaylistManager()
+            result = manager._filter_epg(sample_epg)
+
+            result_str = ET.tostring(result, encoding="unicode")
+
+            # ESPN Sports - starts with ESPN, no Adult - should remain
+            assert "ESPN Sports" in result_str
+            # ESPN Adult - starts with ESPN but contains Adult - should be filtered
+            assert "ESPN Adult" not in result_str
+            # HBO Movies - doesn't start with ESPN - should be filtered
+            assert "HBO Movies" not in result_str
+
+    def test_filter_epg_case_insensitive(self):
+        """Test: _filter_epg es case-insensitive."""
+        sample_epg = b'''<?xml version="1.0" encoding="UTF-8"?>
+        <tv>
+            <channel id="ch1">
+                <display-name>ESPN Sports</display-name>
+            </channel>
+        </tv>'''
+
+        with patch("m3u_list_builder.playlist.settings") as mock_settings:
+            mock_settings.host = "http://test-iptv.com"
+            mock_settings.username = "testuser"
+            mock_settings.password = "testpass"
+            mock_settings.include_text = ["espn"]  # lowercase
+            mock_settings.exclude_text = []
+
+            from m3u_list_builder.playlist import PlaylistManager
+            import xml.etree.ElementTree as ET
+
+            manager = PlaylistManager()
+            result = manager._filter_epg(sample_epg)
+
+            result_str = ET.tostring(result, encoding="unicode")
+
+            # Should still match despite case difference
+            assert "ESPN Sports" in result_str
+
+    # ========================================
+    # Tests para _write_epg
+    # ========================================
+
+    def test_write_epg_creates_file(self, temp_dir):
+        """Test: _write_epg crea el archivo EPG correctamente."""
+        import xml.etree.ElementTree as ET
+
+        with patch("m3u_list_builder.playlist.PUBLIC_DIR", temp_dir):
+            with patch("m3u_list_builder.playlist.settings") as mock_settings:
+                mock_settings.host = "http://test-iptv.com"
+                mock_settings.username = "testuser"
+                mock_settings.password = "testpass"
+
+                from m3u_list_builder.playlist import PlaylistManager
+
+                manager = PlaylistManager()
+
+                root = ET.Element("tv")
+                channel = ET.SubElement(root, "channel", id="ch1")
+                name = ET.SubElement(channel, "display-name")
+                name.text = "Test Channel"
+
+                manager._write_epg(root)
+
+                epg_file = temp_dir / "epg.xml"
+                assert epg_file.exists()
+                content = epg_file.read_text()
+                assert "Test Channel" in content
+
+    def test_write_epg_atomic_replacement(self, temp_dir):
+        """Test: _write_epg reemplaza atómicamente."""
+        import xml.etree.ElementTree as ET
+
+        epg_file = temp_dir / "epg.xml"
+        temp_file = temp_dir / "epg.xml.tmp"
+
+        # Create existing file
+        epg_file.write_text("OLD CONTENT")
+
+        with patch("m3u_list_builder.playlist.PUBLIC_DIR", temp_dir):
+            with patch("m3u_list_builder.playlist.settings") as mock_settings:
+                mock_settings.host = "http://test-iptv.com"
+                mock_settings.username = "testuser"
+                mock_settings.password = "testpass"
+
+                from m3u_list_builder.playlist import PlaylistManager
+
+                manager = PlaylistManager()
+
+                root = ET.Element("tv")
+                manager._write_epg(root)
+
+                # Temp file should not exist after
+                assert not temp_file.exists()
+                # Final file should have new content
+                assert "OLD CONTENT" not in epg_file.read_text()
+
+    def test_fetch_and_generate_calls_epg(self, sample_channels):
+        """Test: fetch_and_generate también llama a fetch_and_write_epg."""
+        with patch("m3u_list_builder.playlist.settings") as mock_settings:
+            mock_settings.host = "http://test-iptv.com"
+            mock_settings.username = "testuser"
+            mock_settings.password = "testpass"
+            mock_settings.output_file = "test_playlist.m3u"
+
+            from m3u_list_builder.playlist import PlaylistManager
+
+            manager = PlaylistManager()
+
+            with patch("m3u_list_builder.playlist.requests.get") as mock_get:
+                mock_response = MagicMock()
+                mock_response.json.return_value = sample_channels
+                mock_response.raise_for_status = MagicMock()
+                mock_get.return_value = mock_response
+
+                with patch.object(manager, "_write_m3u"):
+                    with patch.object(
+                        manager, "fetch_and_write_epg"
+                    ) as mock_epg:
+                        manager.fetch_and_generate()
+
+                        mock_epg.assert_called_once()