new attribute for target, required_epg to remove all live channels without epg, if at least one epg source is defined. (#841)

New Features Added a target option to require valid EPG programme data for live entries. Entries without matching EPG data are filtered out when enabled; non-live entries remain unaffected. Added support for configuring this option in the target editor, with localized labels and explanations. Documentation Documented the new setting, including its default behavior and handling when no EPG source is available.
2026-10-09 01:12:23 +02:00 · 2026-08-29 21:41:38 +02:00
parent ed89117633
commit a683927ece
11 changed files with 545 additions and 73 deletions
@@ -728,16 +728,67 @@ fn assign_epg_icon(
    });
 }

+fn is_live_epg_item(item: &PlaylistItem) -> bool {
+    item.header.xtream_cluster == XtreamCluster::Live && item.header.item_type.is_live()
+}
+
+fn has_processed_epg(item: &PlaylistItem, id_cache: &EpgIdCache) -> bool {
+    item.header.epg_channel_id.as_deref().is_some_and(|id| id_cache.contains_processed_epg_id(id))
+}
+
+fn assign_live_channel_epg(
+    channel: &mut PlaylistItem,
+    id_cache: &EpgIdCache,
+    icon_tags: &HashMap<Arc<str>, &Arc<str>>,
+    icon_override_channels: &HashSet<Arc<str>>,
+    icon_assigned: &mut HashSet<Arc<str>>,
+    stats: &mut EpgAssignmentStats,
+) -> bool {
+    if id_cache.smart_match_enabled {
+        stats.record(assign_smart_epg_id(channel, id_cache));
+    }
+    let has_epg = has_processed_epg(channel, id_cache);
+    assign_epg_icon(channel, icon_tags, icon_override_channels, icon_assigned);
+    has_epg
+}
+
 fn referenced_live_epg_ids(fp: &mut FetchedPlaylist<'_>) -> HashSet<Arc<str>> {
    fp.items()
-        .filter(|channel| channel.header.xtream_cluster == XtreamCluster::Live && channel.header.item_type.is_live())
+        .filter(|channel| is_live_epg_item(channel))
        .filter_map(|channel| {
            channel.header.epg_channel_id.as_ref().map(|id| with_folded_epg_id(id, |folded| folded.intern()))
        })
        .collect()
 }

+pub(crate) fn retain_live_items_with_processed_epg(fp: &mut FetchedPlaylist<'_>, epg: &[Epg]) {
+    let processed_epg_ids = epg
+        .iter()
+        .flat_map(|source| &source.children)
+        .map(|channel| with_folded_epg_id(&channel.id, |folded| folded.intern()))
+        .collect::<HashSet<_>>();
+    let mut removed = 0usize;
+    fp.source.retain_memory_items_mut(|item| {
+        if !is_live_epg_item(item) {
+            return true;
+        }
+        let has_epg = item
+            .header
+            .epg_channel_id
+            .as_deref()
+            .is_some_and(|id| with_folded_epg_id(id, |folded| processed_epg_ids.contains(folded)));
+        if !has_epg {
+            removed += 1;
+        }
+        has_epg
+    });
+    if removed > 0 {
+        debug!("Removed {removed} live channels invalidated by after-EPG mappings from input '{}'", fp.input.name);
+    }
+}
+
 /// Assigns EPG IDs and logos to live playlist channels by matching them with EPG data.
+/// When EPG data is required by the target, unmatched live entries are removed in the same pass.
 ///
 /// For each live channel in the playlist missing an EPG ID, attempts to assign one using normalized name matching if smart matching is enabled. If a channel has an EPG ID but lacks logos, assigns logos from the corresponding EPG icon tags. Adds the matched EPG data to the provided vector.
 ///
@@ -747,68 +798,90 @@ fn referenced_live_epg_ids(fp: &mut FetchedPlaylist<'_>) -> HashSet<Arc<str>> {
 /// let mut new_epg = Vec::new();
 /// let mut playlist = FetchedPlaylist::default();
 /// let mut id_cache = EpgIdCache::new(None);
-/// assign_channel_epg(&mut new_epg, &mut playlist, &mut id_cache);
+/// assign_channel_epg(&mut new_epg, &mut playlist, &mut id_cache, false);
 /// ```
-async fn assign_channel_epg(new_epg: &mut Vec<Epg>, fp: &mut FetchedPlaylist<'_>, id_cache: &mut EpgIdCache) {
-    if let Some(tv_guide) = &fp.epg {
-        if let Some((mut epg_source, icon_override_channels)) =
-            tv_guide.filter_merged_with_icon_overrides(id_cache).await
-        {
-            let stats = {
-                let icon_tags = epg_source
-                    .children
-                    .iter()
-                    .filter_map(|tag| {
-                        tag.icon
-                            .as_ref()
-                            .filter(|icon| !icon.is_empty())
-                            .map(|icon| (with_folded_epg_id(&tag.id, |folded| folded.intern()), icon))
-                    })
-                    .collect::<HashMap<Arc<str>, &Arc<str>>>();
-                let icon_override_channels = icon_override_channels
-                    .into_iter()
-                    .map(|id| with_folded_epg_id(&id, |folded| folded.intern()))
-                    .collect::<HashSet<_>>();
-                let mut icon_assigned = HashSet::new();
-                let mut stats = EpgAssignmentStats::default();
+async fn assign_channel_epg(
+    new_epg: &mut Vec<Epg>,
+    fp: &mut FetchedPlaylist<'_>,
+    id_cache: &mut EpgIdCache,
+    required_epg: bool,
+) {
+    let Some(tv_guide) = &fp.epg else {
+        return;
+    };
+    let mut merged_epg = tv_guide.filter_merged_with_icon_overrides(id_cache).await;
+    if merged_epg.is_none() && !required_epg {
+        return;
+    }

-                if fp.is_memory() {
-                    fp.items_mut()
-                        .filter(|channel| {
-                            channel.header.xtream_cluster == XtreamCluster::Live && channel.header.item_type.is_live()
-                        })
-                        .for_each(|channel| {
-                            if id_cache.smart_match_enabled {
-                                stats.record(assign_smart_epg_id(channel, id_cache));
-                            }
-                            assign_epg_icon(channel, &icon_tags, &icon_override_channels, &mut icon_assigned);
-                        });
-                } else {
-                    warn!("Disk based playlist modification is not supported!");
-                }
-                stats
-            };
+    let mut stats = EpgAssignmentStats::default();
+    let mut removed = 0usize;
+    if fp.is_memory() {
+        let icon_tags = merged_epg
+            .as_ref()
+            .into_iter()
+            .flat_map(|(epg_source, _)| &epg_source.children)
+            .filter_map(|tag| {
+                tag.icon
+                    .as_ref()
+                    .filter(|icon| !icon.is_empty())
+                    .map(|icon| (with_folded_epg_id(&tag.id, |folded| folded.intern()), icon))
+            })
+            .collect::<HashMap<Arc<str>, &Arc<str>>>();
+        let icon_override_channels = merged_epg
+            .as_ref()
+            .into_iter()
+            .flat_map(|(_, channels)| channels)
+            .map(|id| with_folded_epg_id(id, |folded| folded.intern()))
+            .collect::<HashSet<_>>();
+        let mut icon_assigned = HashSet::with_capacity(icon_tags.len());

-            let referenced_epg_ids = referenced_live_epg_ids(fp);
-            epg_source
-                .children
-                .retain(|channel| with_folded_epg_id(&channel.id, |folded| referenced_epg_ids.contains(folded)));
-
-            if id_cache.smart_match_enabled {
-                debug!(
-                    "Smart EPG summary for input '{}': live={}, existing={}, exact={}, fuzzy={}, corrected={}, unresolved={}",
-                    fp.input.name,
-                    stats.live,
-                    stats.existing,
-                    stats.exact,
-                    stats.fuzzy,
-                    stats.corrected,
-                    stats.unresolved
-                );
+        let mut process_channel = |channel: &mut PlaylistItem| {
+            if !is_live_epg_item(channel) {
+                return true;
            }
+            let has_epg = assign_live_channel_epg(
+                channel,
+                id_cache,
+                &icon_tags,
+                &icon_override_channels,
+                &mut icon_assigned,
+                &mut stats,
+            );
+            if required_epg && !has_epg {
+                removed += 1;
+                return false;
+            }
+            true
+        };

-            new_epg.push(epg_source);
+        if required_epg {
+            fp.source.retain_memory_items_mut(&mut process_channel);
+        } else {
+            fp.items_mut().for_each(|channel| {
+                process_channel(channel);
+            });
        }
+    } else {
+        warn!("Disk based playlist modification is not supported!");
+    }
+
+    if id_cache.smart_match_enabled {
+        debug!(
+            "Smart EPG summary for input '{}': live={}, existing={}, exact={}, fuzzy={}, corrected={}, unresolved={}",
+            fp.input.name, stats.live, stats.existing, stats.exact, stats.fuzzy, stats.corrected, stats.unresolved
+        );
+    }
+    if removed > 0 {
+        debug!("Removed {removed} live channels without EPG from input '{}'", fp.input.name);
+    }
+
+    if let Some((mut epg_source, _)) = merged_epg.take() {
+        let referenced_epg_ids = referenced_live_epg_ids(fp);
+        epg_source
+            .children
+            .retain(|channel| with_folded_epg_id(&channel.id, |folded| referenced_epg_ids.contains(folded)));
+        new_epg.push(epg_source);
    }
 }

@@ -821,9 +894,9 @@ async fn assign_channel_epg(new_epg: &mut Vec<Epg>, fp: &mut FetchedPlaylist<'_>
 /// ```text
 /// let mut playlist = FetchedPlaylist::default();
 /// let mut epg_data = Vec::new();
-/// process_playlist_epg(&mut playlist, &mut epg_data);
+/// process_playlist_epg(&mut playlist, &mut epg_data, false);
 /// ```
-pub async fn process_playlist_epg(fp: &mut FetchedPlaylist<'_>, epg: &mut Vec<Epg>) {
+pub async fn process_playlist_epg(fp: &mut FetchedPlaylist<'_>, epg: &mut Vec<Epg>, required_epg: bool) {
    if fp.input.epg.is_none() {
        return;
    }
@@ -831,10 +904,10 @@ pub async fn process_playlist_epg(fp: &mut FetchedPlaylist<'_>, epg: &mut Vec<Ep
    let mut id_cache = EpgIdCache::new(fp.input.epg.as_ref());
    id_cache.collect_epg_id(fp);

-    if id_cache.is_empty() && !id_cache.smart_match_enabled {
+    if id_cache.is_empty() && !id_cache.smart_match_enabled && !required_epg {
        debug!("No epg ids found for input {}", fp.input.name);
    } else {
-        assign_channel_epg(epg, fp, &mut id_cache).await;
+        assign_channel_epg(epg, fp, &mut id_cache, required_epg).await;
    }
 }

@@ -844,7 +917,7 @@ mod tests {
    use rand::{distr::Alphanumeric, Rng};
    use rphonetic::{DoubleMetaphone, Encoder};
    use shared::{
-        model::{ConfigInputDto, PlaylistGroup, PlaylistItemHeader, PlaylistItemType},
+        model::{ConfigInputDto, PlaylistGroup, PlaylistItem, PlaylistItemHeader, PlaylistItemType},
        utils::Internable,
    };
    use std::{collections::HashSet, fs, sync::Arc};
@@ -929,13 +1002,14 @@ mod tests {
        xmltv: &str,
        channels: &[(&str, Option<&str>)],
    ) -> (Vec<Option<Arc<str>>>, Vec<tuliprox_core::model::Epg>) {
-        run_xmltv_matches(xmltv, channels, true).await
+        run_xmltv_matches(xmltv, channels, true, false).await
    }

    async fn run_xmltv_matches(
        xmltv: &str,
        channels: &[(&str, Option<&str>)],
        smart_matching: bool,
+        required_epg: bool,
    ) -> (Vec<Option<Arc<str>>>, Vec<tuliprox_core::model::Epg>) {
        let dir = tempdir().unwrap();
        let epg_path = dir.path().join("smart-match.xml");
@@ -971,7 +1045,7 @@ mod tests {
        };
        let mut epg = Vec::new();

-        super::process_playlist_epg(&mut playlist, &mut epg).await;
+        super::process_playlist_epg(&mut playlist, &mut epg, required_epg).await;
        let assigned_ids = playlist.items_mut().map(|item| item.header.epg_channel_id.clone()).collect();
        (assigned_ids, epg)
    }
@@ -1185,7 +1259,7 @@ mod tests {
            };
            let mut epg = Vec::new();

-            super::process_playlist_epg(&mut playlist, &mut epg).await;
+            super::process_playlist_epg(&mut playlist, &mut epg, false).await;

            let updated = playlist.items_mut().next().unwrap();
            assert_eq!(updated.header.epg_channel_id.as_deref(), Some("demo.channel"));
@@ -1195,6 +1269,211 @@ mod tests {
        });
    }

+    #[test]
+    fn required_epg_removes_only_unmatched_live_items() {
+        let runtime = tokio::runtime::Runtime::new().unwrap();
+        runtime.block_on(async move {
+            let dir = tempdir().unwrap();
+            let epg_path = dir.path().join("required-epg.xml");
+            fs::write(
+                &epg_path,
+                r#"<tv>
+  <channel id="matched.live"><display-name>Matched Live</display-name></channel>
+  <programme start="20260425000000 +0000" stop="20260425010000 +0000" channel="matched.live">
+    <title>Programme</title>
+  </programme>
+</tv>"#,
+            )
+            .unwrap();
+
+            let mut input = ConfigInput::from(ConfigInputDto::default());
+            input.epg = Some(EpgConfig { sources: vec![], smart_match: None });
+            let item = |name: &str, epg_id: Option<&str>, item_type: PlaylistItemType| PlaylistItem {
+                header: PlaylistItemHeader {
+                    name: name.intern(),
+                    epg_channel_id: epg_id.map(Internable::intern),
+                    xtream_cluster: item_type.cluster(),
+                    item_type,
+                    ..PlaylistItemHeader::default()
+                },
+            };
+            let groups = vec![
+                PlaylistGroup {
+                    id: 1,
+                    title: "Live".intern(),
+                    channels: vec![
+                        item("Matched Live", Some("matched.live"), PlaylistItemType::Live),
+                        item("Unmatched Live", Some("missing.live"), PlaylistItemType::LiveHls),
+                    ],
+                    xtream_cluster: super::XtreamCluster::Live,
+                },
+                PlaylistGroup {
+                    id: 2,
+                    title: "Non-Live".intern(),
+                    channels: vec![
+                        item("VOD", None, PlaylistItemType::Video),
+                        item("Series", None, PlaylistItemType::Series),
+                        item("Local VOD", None, PlaylistItemType::LocalVideo),
+                        item("Local Series", None, PlaylistItemType::LocalSeries),
+                    ],
+                    xtream_cluster: super::XtreamCluster::Video,
+                },
+            ];
+            let tv_guide = TVGuide::new(vec![PersistedEpgSource {
+                file_path: epg_path,
+                priority: 0,
+                logo_override: false,
+                kind: PersistedEpgSourceKind::Xmltv,
+            }]);
+            let mut playlist = FetchedPlaylist {
+                input: &input,
+                source: MemoryPlaylistSource::new(groups).into_source(),
+                epg: Some(tv_guide),
+            };
+            let mut epg = Vec::new();
+
+            super::process_playlist_epg(&mut playlist, &mut epg, true).await;
+
+            let names = playlist.items_mut().map(|item| item.header.name.clone()).collect::<HashSet<_>>();
+            assert!(names.contains("Matched Live"));
+            assert!(!names.contains("Unmatched Live"));
+            assert!(names.contains("VOD"));
+            assert!(names.contains("Series"));
+            assert!(names.contains("Local VOD"));
+            assert!(names.contains("Local Series"));
+        });
+    }
+
+    #[test]
+    fn required_epg_is_ignored_without_a_materialized_epg_source() {
+        let runtime = tokio::runtime::Runtime::new().unwrap();
+        runtime.block_on(async move {
+            let mut input = ConfigInput::from(ConfigInputDto::default());
+            input.epg = Some(EpgConfig { sources: vec![], smart_match: None });
+            let groups = vec![PlaylistGroup {
+                id: 1,
+                title: "Live".intern(),
+                channels: vec![live_playlist_item("Unmatched Live", Some("missing.live"))],
+                xtream_cluster: super::XtreamCluster::Live,
+            }];
+            let mut playlist =
+                FetchedPlaylist { input: &input, source: MemoryPlaylistSource::new(groups).into_source(), epg: None };
+
+            super::process_playlist_epg(&mut playlist, &mut Vec::new(), true).await;
+
+            assert_eq!(playlist.items_mut().count(), 1);
+        });
+    }
+
+    #[test]
+    fn optional_epg_preserves_existing_empty_groups() {
+        let runtime = tokio::runtime::Runtime::new().unwrap();
+        runtime.block_on(async move {
+            let dir = tempdir().unwrap();
+            let epg_path = dir.path().join("optional-epg-empty-group.xml");
+            fs::write(
+                &epg_path,
+                r#"<tv>
+  <channel id="matched.live"><display-name>Matched Live</display-name></channel>
+  <programme start="20260425000000 +0000" stop="20260425010000 +0000" channel="matched.live">
+    <title>Programme</title>
+  </programme>
+</tv>"#,
+            )
+            .unwrap();
+
+            let mut input = ConfigInput::from(ConfigInputDto::default());
+            input.epg = Some(EpgConfig { sources: vec![], smart_match: None });
+            let groups = vec![
+                PlaylistGroup {
+                    id: 1,
+                    title: "Empty".intern(),
+                    channels: vec![],
+                    xtream_cluster: super::XtreamCluster::Live,
+                },
+                PlaylistGroup {
+                    id: 2,
+                    title: "Live".intern(),
+                    channels: vec![live_playlist_item("Matched Live", Some("matched.live"))],
+                    xtream_cluster: super::XtreamCluster::Live,
+                },
+            ];
+            let tv_guide = TVGuide::new(vec![PersistedEpgSource {
+                file_path: epg_path,
+                priority: 0,
+                logo_override: false,
+                kind: PersistedEpgSourceKind::Xmltv,
+            }]);
+            let mut playlist = FetchedPlaylist {
+                input: &input,
+                source: MemoryPlaylistSource::new(groups).into_source(),
+                epg: Some(tv_guide),
+            };
+
+            super::process_playlist_epg(&mut playlist, &mut Vec::new(), false).await;
+
+            assert_eq!(playlist.get_group_count(), 2);
+        });
+    }
+
+    #[test]
+    fn required_epg_removes_live_items_without_ids_when_smart_matching_is_disabled() {
+        let runtime = tokio::runtime::Runtime::new().unwrap();
+        runtime.block_on(async move {
+            let dir = tempdir().unwrap();
+            let epg_path = dir.path().join("required-epg-no-ids.xml");
+            fs::write(&epg_path, "<tv></tv>").unwrap();
+
+            let mut input = ConfigInput::from(ConfigInputDto::default());
+            input.epg = Some(EpgConfig { sources: vec![], smart_match: None });
+            let groups = vec![PlaylistGroup {
+                id: 1,
+                title: "Live".intern(),
+                channels: vec![live_playlist_item("Live without ID", None)],
+                xtream_cluster: super::XtreamCluster::Live,
+            }];
+            let tv_guide = TVGuide::new(vec![PersistedEpgSource {
+                file_path: epg_path,
+                priority: 0,
+                logo_override: false,
+                kind: PersistedEpgSourceKind::Xmltv,
+            }]);
+            let mut playlist = FetchedPlaylist {
+                input: &input,
+                source: MemoryPlaylistSource::new(groups).into_source(),
+                epg: Some(tv_guide),
+            };
+
+            super::process_playlist_epg(&mut playlist, &mut Vec::new(), true).await;
+
+            assert_eq!(playlist.items_mut().count(), 0);
+            assert_eq!(playlist.get_group_count(), 0);
+        });
+    }
+
+    #[test]
+    fn required_epg_keeps_live_items_assigned_by_smart_matching() {
+        let runtime = tokio::runtime::Runtime::new().unwrap();
+        runtime.block_on(async move {
+            let (assigned_ids, epg) = run_xmltv_matches(
+                r#"<tv>
+  <channel id="tf1.fr"><display-name>TF1</display-name></channel>
+  <programme start="20260425000000 +0000" stop="20260425010000 +0000" channel="tf1.fr">
+    <title>Programme TF1</title>
+  </programme>
+</tv>"#,
+                &[("TF1", None)],
+                true,
+                true,
+            )
+            .await;
+
+            assert_eq!(assigned_ids, vec![Some("tf1.fr".intern())]);
+            assert_eq!(epg[0].children.len(), 1);
+            assert_eq!(epg[0].children[0].id.as_ref(), "tf1.fr");
+        });
+    }
+
    #[test]
    fn smart_match_replaces_an_existing_id_that_has_no_programmes() {
        let runtime = tokio::runtime::Runtime::new().unwrap();
@@ -1228,7 +1507,7 @@ mod tests {
 </tv>"#;
            for smart_matching in [false, true] {
                let (assigned_ids, epg) =
-                    run_xmltv_matches(xmltv, &[("Empty channel", Some("empty.fr"))], smart_matching).await;
+                    run_xmltv_matches(xmltv, &[("Empty channel", Some("empty.fr"))], smart_matching, false).await;

                assert_eq!(assigned_ids, vec![Some("empty.fr".intern())]);
                assert_eq!(epg[0].children.len(), 1);
@@ -1308,7 +1587,7 @@ mod tests {
            };
            let mut epg = Vec::new();

-            super::process_playlist_epg(&mut playlist, &mut epg).await;
+            super::process_playlist_epg(&mut playlist, &mut epg, false).await;

            assert_eq!(epg.len(), 1);
            assert_eq!(epg[0].children[0].id.as_ref(), "f1.calendar");
@@ -1346,7 +1625,7 @@ mod tests {
            };
            let mut epg = Vec::new();

-            super::process_playlist_epg(&mut playlist, &mut epg).await;
+            super::process_playlist_epg(&mut playlist, &mut epg, false).await;

            let updated = playlist.items_mut().next().unwrap();
            assert_eq!(updated.header.epg_channel_id.as_deref(), Some("f1.calendar"));
@@ -7,8 +7,12 @@ use crate::{
    parser::xmltv::{flatten_tvguide, merge_epg_trees, EpgMergeAccumulator, TVGuide},
    playlist_watch::{process_group_watch, process_target_groups_watch},
    processor::{
-        epg::process_playlist_epg, sort::sort_playlist, trakt::process_trakt_categories_for_target,
-        xtream_series::playlist_resolve_series, xtream_vod::playlist_resolve_vod, StalkerRefreshMode,
+        epg::{process_playlist_epg, retain_live_items_with_processed_epg},
+        sort::sort_playlist,
+        trakt::process_trakt_categories_for_target,
+        xtream_series::playlist_resolve_series,
+        xtream_vod::playlist_resolve_vod,
+        StalkerRefreshMode,
    },
 };
 use futures::{FutureExt, StreamExt};
@@ -21,8 +25,8 @@ use shared::{
    error::{get_errors_notify_message, TuliproxError},
    foundation::{get_field_value, set_field_value, Filter, ValueAccessor, ValueProvider},
    model::{
-        ClusterFlags, CounterModifier, EventMessage, EventSink, FieldGet, FieldSet, InputStats, InputType,
-        MappingStage, PipelineStats, PlaylistGroup, PlaylistItem, PlaylistItemType, PlaylistStats,
+        ClusterFlags, ConfigTargetOptions, CounterModifier, EventMessage, EventSink, FieldGet, FieldSet, InputStats,
+        InputType, MappingStage, PipelineStats, PlaylistGroup, PlaylistItem, PlaylistItemType, PlaylistStats,
        PlaylistUpdateProgressEvent, PlaylistUpdateSummary, ProviderFetchFailure, SourceStats, StreamProperties,
        TargetStats, UUIDType, WatchDisabled, WatchDisabledReason, WatchUnmatched, XtreamCluster,
    },
@@ -1690,7 +1694,9 @@ async fn prepare_playlist_for_target<E: EventSink + Clone + 'static, M: Metadata
        log_memory_snapshot(
            format!("target '{}' input '{}' after_vod_resolve", target.name, provider_fpl.input.name).as_str(),
        );
-        process_playlist_epg(&mut processed_fpl, &mut new_epg).await;
+        let required_epg = target.options.as_ref().is_some_and(ConfigTargetOptions::required_epg);
+        let input_epg_start = new_epg.len();
+        process_playlist_epg(&mut processed_fpl, &mut new_epg, required_epg).await;
        log_memory_snapshot(
            format!("target '{}' input '{}' after_epg_apply", target.name, processed_fpl.input.name).as_str(),
        );
@@ -1702,6 +1708,9 @@ async fn prepare_playlist_for_target<E: EventSink + Clone + 'static, M: Metadata
            deduplicate.then_some(&mut duplicates),
        ) {
            processed_fpl.source = MemoryPlaylistSource::new(groups).into_source();
+            if required_epg && new_epg.len() > input_epg_start {
+                retain_live_items_with_processed_epg(&mut processed_fpl, &new_epg[input_epg_start..]);
+            }
        }
        if let Some(stat) = stats.get_mut(&processed_fpl.input.name) {
            stat.processed_stats.group_count = processed_fpl.get_group_count();
@@ -3614,6 +3623,82 @@ match {
            });
        }

+        #[test]
+        fn required_epg_removes_ids_invalidated_by_after_epg_mapping() {
+            let runtime = Runtime::new().expect("runtime");
+            runtime.block_on(async {
+                let dir = tempdir().expect("tempdir");
+                let ics_path = dir.path().join("bbc.ics");
+                std::fs::write(
+                    &ics_path,
+                    "BEGIN:VCALENDAR\nBEGIN:VEVENT\nSUMMARY:News\nDTSTART:20260306T120000Z\nDTEND:20260306T130000Z\nEND:VEVENT\nEND:VCALENDAR",
+                )
+                .expect("write ics");
+
+                let mut input = ConfigInput::from(ConfigInputDto::default());
+                input.name = "input".intern();
+                input.epg = Some(EpgConfig { sources: vec![], smart_match: None });
+                let groups = vec![PlaylistGroup {
+                    id: 1,
+                    title: "Live".intern(),
+                    channels: vec![PlaylistItem {
+                        header: PlaylistItemHeader {
+                            name: "BBC One".intern(),
+                            epg_channel_id: Some("bbc.one".intern()),
+                            group: "Live".intern(),
+                            xtream_cluster: XtreamCluster::Live,
+                            item_type: PlaylistItemType::Live,
+                            ..Default::default()
+                        },
+                    }],
+                    xtream_cluster: XtreamCluster::Live,
+                }];
+                let tv_guide = TVGuide::new(vec![PersistedEpgSource {
+                    file_path: ics_path,
+                    priority: 0,
+                    logo_override: false,
+                    kind: PersistedEpgSourceKind::Ics {
+                        channel_id: "bbc.one".intern(),
+                        channel_title: Some("BBC One".intern()),
+                        match_names: vec![],
+                        config: Box::new(IcsEpgSourceConfig::default()),
+                    },
+                }]);
+                let mut playlist = FetchedPlaylist {
+                    input: &input,
+                    source: MemoryPlaylistSource::new(groups).into_source(),
+                    epg: Some(tv_guide),
+                };
+
+                let rewrite_epg =
+                    build_mapping("rewrite", MappingStage::AfterEpg, r#"@epg_channel_id = "missing.epg""#);
+                let add_virtual = build_mapping("virtual", MappingStage::AfterEpg, r#"add_favourite("Echo")"#);
+                let mut target = build_target(vec![rewrite_epg, add_virtual], false);
+                target.options = Some(ConfigTargetOptions { required_epg: true, ..Default::default() });
+                let mut stats = HashMap::from([(
+                    Arc::clone(&input.name),
+                    create_input_stat(1, 1, 0, input.input_type, &input.name, 0),
+                )]);
+                let mut errors = Vec::new();
+
+                let prepared = prepare_playlist_for_target(
+                    &processing_context(),
+                    std::slice::from_mut(&mut playlist),
+                    &target,
+                    &mut stats,
+                    &mut errors,
+                    false,
+                )
+                .await
+                .expect("target preparation");
+
+                assert!(errors.is_empty());
+                assert!(prepared.playlist.is_empty());
+                assert_eq!(stats[&input.name].processed_stats.group_count, 0);
+                assert_eq!(stats[&input.name].processed_stats.channel_count, 0);
+            });
+        }
+
        fn live_item_for_epg(name: &str) -> PlaylistItem {
            PlaylistItem {
                header: PlaylistItemHeader {