Skip to content

No Match #6

@cammi88

Description

@cammi88

i just tried to use the script and git no match error.
claude fixed this and told me it had to do with line "downloaded = 1"

this is the script it created for me and it worked. maybe this helps someone someday

#!/usr/bin/env python3

import sys
import zipfile
import sqlite3
from dataclasses import dataclass
from operator import itemgetter
from pathlib import Path
from sqlite3 import Cursor

from matcher import ObjectListMatcher



CUR_PATH = Path()

EPISODES_DIR_PATH = '/storage/emulated/0/Android/data/de.danoeh.antennapod/files/media/from_podcast_addict'
MATCH_ON_EPISODE_URL_IF_COULD_NOT_FIND_A_MATCH_OTHERWISE = True


@dataclass
class Feed:
    id: int
    name: str
    description: str
    author: str
    keep_updated: int
    folder_name: str = ''


def error(msg):
    print("ERROR:", msg)
    sys.exit(1)


def confirmed(msg):
    inp = input(msg + " (y = yes)\n> ").strip().lower()
    return inp in ('y', 'yes')


def get_one_file_or_error(glob_pattern, path=CUR_PATH):
    files = list(path.glob(glob_pattern))
    if not files:
        error("found no file for the pattern: " + glob_pattern)
    if len(files) > 1:
        error("found more than one file for pattern: " + glob_pattern)
    return files[0]


def get_antenna_pod_and_podcast_addict_backup_path():
    podcast_addict_backup_file = get_one_file_or_error("PodcastAddict*.backup")
    antenna_pod_db_file = get_one_file_or_error("AntennaPodBackup*.db")

    path_to_extract_to = CUR_PATH / 'podcast_addict_extracted'

    podcast_addict_db_file = next(path_to_extract_to.glob("*.db"), None)
    if podcast_addict_db_file is None:
        path_to_extract_to.mkdir(exist_ok=True)

        if not zipfile.is_zipfile(podcast_addict_backup_file):
            error("somehow the Podcast Addict .backup file is not a zip")

        z = zipfile.ZipFile(podcast_addict_backup_file)

        print("Extracting Podcast Addict backup..")
        z.extractall(path=path_to_extract_to)

        podcast_addict_db_file = get_one_file_or_error(  #
                "*.db", path=path_to_extract_to)

    return antenna_pod_db_file, podcast_addict_db_file



def transfer(podcast_addict_cur: Cursor, antenna_pod_cur: Cursor):
    # first find match for all feeds in pa
    pa_feeds = [Feed(*a) for a in podcast_addict_cur.execute(
            'select _id, name, description, author, '
            'automaticRefresh, folderName from podcasts '
            'where subscribed_status = 1 and is_virtual = 0')]

    print("# Podcast addict feeds:")
    for feed in pa_feeds:
        print(feed.name)
    print()
    print()

    ap_feeds = [Feed(*a) for a in antenna_pod_cur.execute(
            'select id, title, description, author, keep_updated from Feeds '
            'where downloaded != 0')]  # FIX: newer AP versions store a timestamp instead of 1

    feed_attr_to_weight = {  #
        (lambda f: f.name): 0.85,  #
        (lambda f: f.author): 0.15,  #
    }
    matcher = ObjectListMatcher(feed_attr_to_weight)

    # should never be larger than the largest weight (otherwise is
    # slightly unpredictable, as not every weight will be evaluated)
    # value in range [0, 1]
    matcher.minimum_similarity = 0.78

    pa_to_ap = []

    ap_indices = matcher.get_indices(pa_feeds, ap_feeds)
    for n, pa in enumerate(pa_feeds):
        ap_idx = ap_indices[n]

        ap_name = '!!! NO MATCH !!!'
        if ap_idx >= 0:
            ap = ap_feeds[ap_idx]
            ap_name = ap.name
            pa_to_ap.append((pa, ap))

        print(pa.name, ap_name, sep="  ->  ")
    print()

    if not confirmed("Is this correct? Can we continue?"):
        return

    # if you want to merge podcasts (e.g. non-premium and premium -> premium)
    #for pa in pa_feeds:
    #    if pa.name == "Name of non premium podcast feed":
    #        for ap in ap_feeds:
    #            # FIXME: make it work if premium and non-premium share same name
    #            if ap.name == "Name of same podcast but premium version":
    #                transfer_from_feed_to_feed(podcast_addict_cur,
    #                                           antenna_pod_cur, pa, ap)
    #                break
    #        break


    for pa, ap in pa_to_ap:
        transfer_from_feed_to_feed(podcast_addict_cur, antenna_pod_cur, pa, ap)
        print()  # break


ITEM_MATCHER = ObjectListMatcher({(lambda i: i[1]): 1})
ITEM_MATCHER.minimum_similarity = 0.83
ITEM_MATCHER.lock_in_if_similarity_first_above = 0.97


def transfer_from_feed_to_feed(podcast_addict_cur: Cursor,  #
                               antenna_pod_cur: Cursor,  #
                               pa: Feed,  #
                               ap: Feed):
    print(f'# Feed: {ap.name}')
    antenna_pod_cur.execute("UPDATE Feeds "
                            "SET keep_updated = ? "
                            "WHERE id = ?",  #
                            (pa.keep_updated, ap.id,))

    pa_episodes = list(podcast_addict_cur.execute(  #
            #        0   1     n2            n3       n4
            'select _id, name, seen_status, favorite, local_file_name, '
            # n5           n6           n7                  n8
            'playbackDate, duration_ms, chapters_extracted, download_url '
            'from episodes where podcast_id = ? and '
            '(seen_status = 1 or '
            '(local_file_name != "" and local_file_name IS NOT NULL))',
            (pa.id,)))

    ap_episodes = list(antenna_pod_cur.execute(  #
            'select fi.id, fi.title, fm.download_url '
            'from FeedItems fi '
            'LEFT JOIN FeedMedia fm ON fi.id = fm.feeditem '
            'where fi.feed = ? and fi.read = 0 ', (ap.id,)))

    print()
    combinations = len(pa_episodes) * len(ap_episodes)
    print(f"Rough estimate: {combinations / 4000:.2f} seconds")
    print()
    print()
    pa_indices = ITEM_MATCHER.get_indices(ap_episodes, pa_episodes)
    seen_match_count = 0


    for ap_ep, pa_idx in zip(ap_episodes, pa_indices):
        if pa_idx < 0:
            found = False

            # give it one last chance
            ap_url = ap_ep[2]
            if MATCH_ON_EPISODE_URL_IF_COULD_NOT_FIND_A_MATCH_OTHERWISE and ap_url is not None:
                ap_url = ap_url.strip()
                if len(ap_url) > 9:
                    for pa_idx, pa_ep in enumerate(pa_episodes):
                        if not pa_ep[8]:
                            continue

                        pa_url = pa_ep[8].strip()
                        if pa_url and pa_url == ap_url:
                            found = True
                            break

            if not found:
                print(f"!  No match for: {ap_ep[1]}")
                continue

        seen_match_count += 1
        pa_ep = pa_episodes[pa_idx]
        if pa_ep[2]:
            transfer_from_seen_ep_to_ep(antenna_pod_cur, podcast_addict_cur,  #
                                        pa_ep, ap_ep)

        if pa_ep[3]:
            antenna_pod_cur.execute(
                "INSERT INTO Favorites (feeditem, feed) VALUES "
                "(?, ?)", (ap_ep[0], ap.id))

        if pa_ep[4]:
            transfer_from_dld_ep_to_ep(antenna_pod_cur, podcast_addict_cur,  #
                                       pa_ep, ap_ep, pa.folder_name)

        if pa_ep[7]:
            transfer_chapters(antenna_pod_cur, podcast_addict_cur,  #
                              pa_ep, ap_ep, pa.id)

    print(f'\nINFO: In this feed {seen_match_count} episodes were matched')


def transfer_chapters(antenna_pod_cur: Cursor,  #
                      podcast_addict_cur: Cursor,  #
                      pa_ep: tuple,  #
                      ap_ep: tuple, pa_feed_id: int):
    for title, start in podcast_addict_cur.execute(  #
            "select name, start from chapters "
            "where podcastId = ? and episodeId = ?", (pa_feed_id, pa_ep[0])):
        # we use chapter type 2 (id3) simply because it seems most likely
        antenna_pod_cur.execute("INSERT INTO SimpleChapters "
                                "(title, start, feeditem, type) VALUES "
                                "(?, ?, ?, 2)", (title, start, ap_ep[0],))


def transfer_from_dld_ep_to_ep(antenna_pod_cur: Cursor,  #
                               podcast_addict_cur: Cursor,  #
                               pa_ep: tuple,  #
                               ap_ep: tuple,  #
                               pa_folder_name: str):
    pa_ep_id, _, _, _, pa_local_file_name, _, _, _, _ = pa_ep

    dir_path = EPISODES_DIR_PATH.rstrip("/") + "/" + pa_folder_name
    file_path = dir_path + "/" + pa_local_file_name
    antenna_pod_cur.execute("UPDATE FeedMedia "
                            "SET file_url = ?, "
                            "downloaded = 1 "
                            "WHERE feeditem = ?",  #
                            (file_path, ap_ep[0],))


def transfer_from_seen_ep_to_ep(antenna_pod_cur: Cursor,  #
                                podcast_addict_cur: Cursor,  #
                                pa_ep: tuple,  #
                                ap_ep: tuple):
    print(ap_ep[1], "  <<matched to>>  ", pa_ep[1])
    pa_ep_id, _, _, _, _, pa_playbackDate, pa_duration_ms, _, _ = pa_ep
    antenna_pod_cur.execute("UPDATE FeedItems SET read = 1 WHERE id = ?",
                            (ap_ep[0],))

    antenna_pod_cur.execute("UPDATE FeedMedia "
                            "SET playback_completion_date = ?, "
                            "last_played_time = ?, "
                            "played_duration = ? "
                            "WHERE feeditem = ?",  #
                            (pa_playbackDate, pa_playbackDate, pa_duration_ms,
                             ap_ep[0],))


ap_db, pa_db = get_antenna_pod_and_podcast_addict_backup_path()
print()
print("AntennaPod .db file found:", ap_db)
print("Podcast Addict .db file found:", pa_db)
print()
print()

podcast_addict_con = None
antenna_pod_con = None
try:
    podcast_addict_con = sqlite3.connect(pa_db)
    antenna_pod_con = sqlite3.connect(ap_db)

    transfer(podcast_addict_con.cursor(), antenna_pod_con.cursor())
finally:
    antenna_pod_con.commit()

    if podcast_addict_con is not None:
        antenna_pod_con.close()

    if antenna_pod_con is not None:
        podcast_addict_con.close()

Metadata

Metadata

Assignees

No one assigned

    Labels

    No labels
    No labels

    Type

    No type

    Projects

    No projects

    Milestone

    No milestone

    Relationships

    None yet

    Development

    No branches or pull requests

    Issue actions