用SQLite打造IPTV源数据库：从M3U清洗到可用性检测

简介：IPTV+数据库.rar 是一份融合IPTV系统与数据库应用的学习研究材料，主要面向IPTV业务开发、运维人员及对电视后台技术感兴趣的初学者。压缩包内包含完整的Web服务端与前端代码，其中PHP文件负责后台业务逻辑，JS脚本处理页面交互与数据请求，CSS/HTML构建界面展示，SQL、DB及DAT文件则对应数据库结构与业务数据，可据此了解IPTV系统中用户管理、内容管理、节目单存储等模块的表设计与读写方式。包体共362个文件，含APK客户端安装包、字体图标资源等，整体24.92MB，目录层次清晰。目前已有172人学习下载。全部内容由互联网收集整理，仅供个人学习与文献参考，请勿用于商业用途，并请在下载后24小时内删除。整体适合在本地环境开展功能测试与代码走读，能直观展示IPTV系统前后端交互及数据流转的实现路径。

1. 拿到 "IPTV+数据库.rar"：先搞清这个包到底能解决什么问题

把 rar 解压开，里面不是一沓散乱的 m3u，而是一个组合套件：播放列表文本、建表 SQL、解析入库脚本。在那之前我一直用文本播放列表直接维护，几百个频道找失效源全靠播放器报错，属于典型的人肉运维。

这个包想解决的问题很直接：让 IPTV 源数据从"文本"变成"数据库记录"，于是可以按频道查、按分类统计、给源打可用性标记，再基于结果重新导出干净的列表。它适合维护过播放列表、被重复源和失效源折磨过的开发者。

本文不讨论源列表怎么获得，只讲拿到手之后：怎么拆包、建表、清洗、入库、避坑、再导出。

2. 拆开 RAR 看数据结构：播放列表文件到底是什么形态

2.1 解压命令与结构确认：先看包里到底是啥

拿到任何.rar，第一步永远是先解压到独立目录，而不是直接在压缩包里双击预览。RAR 本质是传输封装，不是数据库备份，里面文件的真实形态往往和文件名暗示的不一致——可能有一个 M3U，也可能有几十个分类文件和一个说明文档。

# 解压到单独目录，避免文件和源码混在一起 mkdir -p iptv_db_project && unrar x "IPTV+数据库.rar" -d iptv_db_project # 如果系统没装 unrar，7z 也能处理 rar 格式 7z x "IPTV+数据库.rar" -oiptv_db_project # 解压后看一眼实际内容 ls -lh iptv_db_project

我这里优先用unrar x是因为它会把完整路径结构带出来，e参数只解文件不保留目录，遇到同名文件会互相覆盖。ls -lh能顺便看到每个文件大小——如果某个 m3u 只有几百字节，它大概率是空模板；如果 SQL 文件比文本还大，说明作者把预聚合结果一起打了包。

解压后建议再检查一下文本文件的行尾符和编码，这一步直接影响后面的解析脚本能不能一次跑通。常见的播放列表文件有两种来源：Windows 下生成的CRLF行尾，Linux 服务器上保存的LF行尾。解析脚本如果按\n切行，CRLF 会被带进地址行最后，导致源地址里藏着\r。

提示：不要上来就用 Python 直接读文件。先用系统命令确认形态，再决定解析策略，这几十秒能省掉后面半小时的排错。

2.2 播放列表的三种常见格式：M3U、纯文本、JSON 怎么选

现实中汇总过来的播放列表主要有三种形态，选型决定了入库脚本的复杂度。

我一般会先把所有文件都尝试按 M3U 解析，解析失败的通道落到纯文本逻辑。原因很实际：M3U 的#EXTINF行虽然乱，但里面往往带着group-title和tvg-logo，这两个字段入库后就是现成的分类和图标；纯文本行入库后只能靠分辨率或命名规律做弱推断。

JSON 格式反而容易踩坑。很多 JSON 播放列表里的字段名不是标准命名，有叫name、也有叫title、channel的，地址字段可能叫url也可能叫streamUrl。入库前要做一层字段映射，否则查出来的数据缺胳膊少腿。

2.3 先做裸文件体检：编码、行尾符和字段抽取

跑任何解析脚本前，先对原始文件做一次 "裸文件体检"。这一步不碰数据，只是拿系统工具看文件的物理属性：

# 查看编码类型和行尾符 file ch_list.m3u # 看开头几行，确认是否有 EXTM3U 头、字段风格 head -n 20 ch_list.m3u # 如果有 cat -A，还能看到行尾是 $ 还是 ^M$ cat -A ch_list.m3u | head -n 5

file输出里如果写着Non-ISO extended-ASCII或ISO-8859，基本可以断定文件不是 UTF-8，大概率是 GBK 或 GB18030 编码。这时强行用utf-8解码会在中文字符处抛异常，或者更恶劣——不抛异常但输出一堆乱码。head是为了确认字段风格：合法的 M3U 第一行是#EXTM3U，随后是#EXTINF与地址交替；如果开头直接就是http://，说明这文件是被精简过的纯文本列表，解析策略要换。

cat -A看行尾是我的习惯动作。行尾出现^M$意味着是 Windows 的 CRLF，Python 的strip()能处理，但如果有人用split(',')直接切地址行，最后一个字段可能带着\r。与其在脚本里做一堆防御，不如导入前先统一转换一次。

做完体检，心里就有谱了：文件是什么编码、行尾是什么风格、字段完整度如何。接下来可以动手建表。

3. 设计频道与源的库表结构：用 SQLite 建表并写解析入库脚本

3.1 三张表的设计：频道、源、检测记录怎么分工

第一次做这个方案的人最容易犯的错，是把所有字段塞进一张大宽表：频道名、分组、地址、状态全放一行。短时间用着没问题，但一个频道经常对应多条源，分组里还可以有多个频道，一张表做下来全是重复数据。

我采用的方案是拆三张表：channel存频道主体信息，stream_source存每个频道的可用源地址，source_check存每次检测结果。频道和源是一对多关系，检测记录和源也是一对多关系，这样历史检测数据能滚动保留，不会覆盖掉上一次的体检结果。

-- 频道表：一个频道只有一行，name 做唯一键 CREATE TABLE IF NOT EXISTS channel ( id INTEGER PRIMARY KEY AUTOINCREMENT, name TEXT NOT NULL UNIQUE, group_name TEXT DEFAULT '', logo TEXT DEFAULT '', created_at TEXT DEFAULT (datetime('now', 'localtime')) ); -- 源表：同一频道同一地址只允许一条记录 CREATE TABLE IF NOT EXISTS stream_source ( id INTEGER PRIMARY KEY AUTOINCREMENT, channel_id INTEGER NOT NULL REFERENCES channel(id), url TEXT NOT NULL, resolution TEXT DEFAULT '', status TEXT DEFAULT 'unknown', checked_at TEXT, created_at TEXT DEFAULT (datetime('now', 'localtime')), UNIQUE(channel_id, url) ); -- 检测记录表：每次体检追加一行，保留历史 CREATE TABLE IF NOT EXISTS source_check ( id INTEGER PRIMARY KEY AUTOINCREMENT, source_id INTEGER NOT NULL REFERENCES stream_source(id), status TEXT NOT NULL, latency_ms INTEGER DEFAULT 0, checked_at TEXT DEFAULT (datetime('now', 'localtime')) ); CREATE INDEX IF NOT EXISTS idx_source_status ON stream_source(status);

字段设计上有两个关键点。stream_source的UNIQUE(channel_id, url)是硬去重，哪怕同一个地址在原始列表里出现五次，入库时也只落一条。status字段默认unknown，后续检测脚本会把它更新为ok、timeout或refused，这样导出干净列表时直接用WHERE status='ok'过滤。

检测记录表为什么单独建？因为检测状态是时序数据，这次通了下次可能断。原始源表只保存当前状态，检测表保存每次体检的历史，以后想统计"这个源一周内稳定率多少"就有据可查。

3.2 解析入库脚本：把 M3U 文本吃进 SQLite

表结构建好，接下来写解析脚本。核心逻辑是逐行扫描 M3U 文件：遇到#EXTINF行就解析出频道名、分组、图标，遇到普通行就当成源地址插入。

import sqlite3 import re from pathlib import Path M3U_FILE = Path('ch_list.m3u') db = sqlite3.connect('iptv.db') db.executescript(""" PRAGMA journal_mode=WAL; PRAGMA foreign_keys=ON; """) current_channel_id = None def parse_extinf(line: str): """从 #EXTINF 行抽取分组、图标和频道名""" group = '' logo = '' m = re.search(r'group-title="([^"]*)"', line) if m: group = m.group(1).strip() m = re.search(r'tvg-logo="([^"]*)"', line) if m: logo = m.group(1).strip() # 频道名在最后一个逗号之后 title = line.split(',', 1)[1].strip() if ',' in line else '' return group, logo, title with open(M3U_FILE, 'r', encoding='utf-8', errors='replace') as f: for raw_line in f: line = raw_line.strip() if not line: continue if line.startswith('#EXTM3U'): continue if line.startswith('#EXTINF'): group, logo, title = parse_extinf(line) if not title: continue # 频道重复时保留已有 id，只更新分组信息 db.execute( "INSERT INTO channel(name, group_name, logo) VALUES(?,?,?) " "ON CONFLICT(name) DO UPDATE SET group_name=excluded.group_name", (title, group, logo) ) row = db.execute("SELECT id FROM channel WHERE name=?", (title,)).fetchone() current_channel_id = row[0] elif line.startswith('#'): continue else: if current_channel_id is None: continue db.execute( "INSERT INTO stream_source(channel_id, url) VALUES(?,?) " "ON CONFLICT(channel_id, url) DO NOTHING", (current_channel_id, line) ) db.commit() print("channels:", db.execute("SELECT COUNT(*) FROM channel").fetchone()[0]) print("sources:", db.execute("SELECT COUNT(*) FROM stream_source").fetchone()[0])

脚本里有几个容易被忽略的点。encoding='utf-8', errors='replace'是故意写的宽松策略：文件里混着 GBK 字符时，不会让整个脚本崩溃，把坏字符替换成占位符，后面清洗再处理。ON CONFLICT是幂等插入的关键，重复执行脚本不会产生重复数据。current_channel_id记录上一次#EXTINF对应的频道 id，保证地址行和频道正确关联。

大量数据入库不要一行一提交。脚本全部插入完后一次性commit()，SQLite 事务开销能被吃掉大半；如果文件有十几万行，可以每 1000 行commit一次，控制内存占用。

3.3 边界情况：duration、空行和多地址行怎么处理

播放列表文件从来不是规规矩矩的。最常见的干扰是#EXTINF行里的时长字段：标准 M3U 里是#EXTINF:-1，但有些文件会写成#EXTINF:0或者直接缺省。解析时不要对-1做特殊判断，统一忽略前半段，只取逗号后的频道名。

空行和注释行好处理，strip()后跳过即可。真正麻烦的是"一个频道后面跟了多个地址行"——这在聚合型源列表里非常常见，一个频道名称下面挂了四五个不同域名的地址。脚本逻辑天然支持这种情况：每个地址都会插入新的stream_source记录，除非完全重复。

还有一类来源会在地址行前面加#注释，比如# 备用地址，随后另起一行写真正的 URL。这类注释行会被当成普通地址入库，检测阶段全部超时。所以入库后一定要跑一遍来源体检，而不是入库完就高枕无忧。

注意：errors='replace'虽然避免了崩溃，但会在中文位置留下 Unicode 替换字符。如果发现频道名里有大量乱码，回到 5.1 节用编码探测重新解析。

4. 数据库里的源体检：用 SQL 统计去重并标记失效源

4.1 用 SQL 做频道聚合：重复源和分类数量一目了然

入库之后最值钱的一个动作，是把散乱的记录变成统计视图。先跑两个最基础的聚合查询，确认这批数据的整体面貌：

-- 按分类统计频道数量，看分组名称是否规范 SELECT group_name, COUNT(*) AS cnt FROM channel GROUP BY group_name ORDER BY cnt DESC; -- 抓出源数量异常多的频道，大概率是重复聚合 SELECT c.name, COUNT(s.id) AS src_cnt FROM channel c JOIN stream_source s ON s.channel_id = c.id GROUP BY c.id HAVING src_cnt > 5 ORDER BY src_cnt DESC LIMIT 20;

第一个查询能暴露分组命名混乱的问题。同一个分类可能叫"央视"也叫"CCTV"，甚至"央视频道"混在一起，后续重新导出时无法按分类合并。第二个查询直接命中重复源——正常一个频道保留 3 到 5 个备用源就够，出现十几个源说明原始列表做了多轮聚合。

这两条 SQL 是所有后续清洗动作的依据：先看整体分布，再定位具体异常记录。不要一上来就删数据，先统计，再决定。

4.2 给源打可用性标记：检测记录表完成一次体检

统计只能看结构，源能不能播需要实际探测。常见做法是并发对每个 URL 发起请求，按响应状态和耗时打标。这里只做最轻量的 TCP 握手检测，不拉取完整播放流，否则几百个源测一轮要等很久。

import socket import sqlite3 from concurrent.futures import ThreadPoolExecutor def check_url(url: str, timeout: float = 4.0) -> str: """对 http/https 源做 TCP 连通性检测""" try: host_port = url.split('://')[1].split('/')[0] if ':' in host_port: host, port = host_port.rsplit(':', 1) port = int(port) else: host, port = host_port, 80 with socket.create_connection((host, port), timeout=timeout): return 'ok' except Exception: return 'timeout' db = sqlite3.connect('iptv.db', timeout=10) sources = db.execute( "SELECT id, url FROM stream_source WHERE status != 'ok'" ).fetchall() with ThreadPoolExecutor(max_workers=16) as pool: results = list(pool.map(lambda s: (s[0], check_url(s[1])), sources)) for source_id, status in results: db.execute( "UPDATE stream_source SET status=?, checked_at=datetime('now','localtime') WHERE id=?", (status, source_id) ) db.execute( "INSERT INTO source_check(source_id, status) VALUES(?,?)", (source_id, status) ) db.commit()

ThreadPoolExecutor(max_workers=16)是实用设置：16 个并发足够快，又不至于把路由器打满。timeout=4秒是平衡值，低于 2 秒误杀率高，高于 6 秒检测一轮太久，遇到电信运营商对境外地址超时的情况，4 秒能让检测稳定许多。

注意：这个检测脚本只适合http://或https://开头、带 CF 域名的源。如果原始列表里有rtsp://或纯 IP 地址，TCP 握手逻辑不一定适用，最好单独写一个基于urllib的检测分支。

4.3 统计结果怎么反推清洗规则

体检完成后，再跑一次按状态分组的统计，清洗规则基本就出来了：

-- 按状态统计，看可用源占比 SELECT status, COUNT(*) FROM stream_source GROUP BY status; -- 找完全没有任何可用源的频道 SELECT c.name, COUNT(s.id) AS total_src FROM channel c LEFT JOIN stream_source s ON s.channel_id = c.id AND s.status = 'ok' GROUP BY c.id HAVING total_src IS NULL OR total_src = 0;

可用率低于 60% 的频道，说明这批源质量整体偏差，优先考虑从其他频道补源或者直接下线。一个源如果有超过三次检测记录且全部超时，那基本可以判定是死链——但别立刻删除，先保留一段时间，因为有些源只在特定时段有效，存在"白天通晚上断"的情况。

我的习惯是保留死链记录但不让它进入导出结果。导出时用WHERE status='ok'过滤，检测失败的历史数据留着做稳定率分析。

5. IPTV 数据入库的四个常见坑：编码、截断、重复与文件锁

5.1 中文频道名乱码：先探测编码再入库

现象：入库后查出来的频道名是���，或者变成一长串类似CCTV-1涓栫晫的内容。

原因：播放列表文件实际是 GBK/GB18030 编码，脚本却按 UTF-8 解码。更隐蔽的情况是文件前半段是 UTF-8、后半段混入 GBK 片段，单次解码必然出问题。

解决：写脚本之前先跑file确认编码，GBK 文件就用gb18030解码。实在无法确认时，写一个探测函数：

def detect_encoding(path: str) -> str: raw = open(path, 'rb').read() # UTF-8 BOM 优先 if raw.startswith(b'\xef\xbb\xbf'): return 'utf-8-sig' try: raw.decode('utf-8') return 'utf-8' except UnicodeDecodeError: return 'gb18030'

gb18030能覆盖 GBK 和 GB2312 全部字符，比gbk编码适用范围更广。这个函数挂在解析脚本最前面，比我当年手动改编码再重跑高效得多。

5.2 频道名被截断：不要把整行按逗号无脑 split

现象：入库后频道名叫"CCTV"或者"新闻"，完整的"CCTV-1 综合"和"新闻综合"被截掉了。

原因：#EXTINF行格式是属性在前、名称在后，中间以逗号分隔。有人直接line.split(',')然后取第一段，结果拿到的是#EXTINF:-1 tvg-id="..."整串，频道名反而丢了。更常见的错误是有人想先剥属性，用split(',')把行切成多段，误把"div"这类逗号前后的内容当成了频道名。

解决：用line.split(',', 1)只切第一个逗号，后面的内容全部保留作为频道名。这样即使频道名本身带逗号，也不会被截断。

# 错误写法：频道名被截断 # title = line.split(',')[1] # 正确写法：只切第一个逗号 title = line.split(',', 1)[1].strip()

5.3 同一频道重复几百条：先聚类再去重

现象：按频道名查询时，一个频道有几十条源记录，其中绝大多数是同一个域名的不同路径，或者同一个源被三个不同分类重复收录。

原因：聚合型源列表本身就是多个人维护的产物，A 收了某卫视、B 也收了，合并时没人做规范化。完全 URL 相同的重复会被UNIQUE索引挡住，但"同一个流媒体的不同 CDN 地址"拦不住。

解决：入库后做一次归一化聚类。把 URL 中的时间参数、随机参数去掉，提取"协议+域名+路径主体"做聚合键，超过 20 条记录的频道只保留来源域名不同的前 5 条，其余标记为duplicate。

-- 找出靠参数区分的重复地址 SELECT url, REPLACE(url, '?', ' ') AS url_key FROM stream_source WHERE status = 'ok' ORDER BY url_key;

这个操作不要写进主流程，跑一次生成对照表，人工确认之后再删。

5.4 数据库文件被锁定：换 WAL 模式，读写分离

现象：导出脚本运行时报database is locked，或者检测脚本连续更新几百条记录后突然报错。

原因：SQLite 默认的 journal 模式在写事务期间会对整个数据库文件加锁，而检测脚本是"边读边写、持续多轮"，每轮更新都可能和读取线程撞锁。更直接的场景是播放器工具直接打开iptv.db文件，锁没释放，脚本写入必然失败。

解决：在链接时强制开启 WAL 模式，同时把读写连接分开。WAL 模式下读操作不会被写操作阻塞，这是 SQLite 并发效率提升最明显的一个开关。

db = sqlite3.connect('iptv.db') db.execute("PRAGMA journal_mode=WAL;") db.execute("PRAGMA busy_timeout=5000;")

busy_timeout=5000是另一个实用参数：碰到锁时等 5 秒而不是立刻失败，在检测脚本并发更新时能大幅减少重试逻辑。

6. 把 SQLite 导出回 M3U：这套结构能不能顶替手工维护

6.1 导出脚本：查询、排序、拼接一次完成

数据库建好、体检做完，最后一个动作是把数据导出成干净的 M3U。导出脚本要解决三个问题：过滤状态、分组排序、拼接格式。

import sqlite3 db = sqlite3.connect('iptv.db') db.row_factory = sqlite3.Row # 只导出状态为 ok 的源，按分组和频道名排序 rows = db.execute(""" SELECT c.name, c.group_name, c.logo, s.url FROM channel c JOIN stream_source s ON s.channel_id = c.id WHERE s.status = 'ok' ORDER BY c.group_name, c.name, s.id """).fetchall() with open('clean.m3u', 'w', encoding='utf-8') as f: f.write('#EXTM3U\n') for row in rows: logo_attr = f' tvg-logo="{row["logo"]}"' if row["logo"] else '' f.write(f'#EXTINF:-1 group-title="{row["group_name"]}"{logo_attr},{row["name"]}\n') f.write(f'{row["url"]}\n') print(f'exported {len(rows)} entries')

group-title如果原数据里是空的，导出时仍然保留空字符串，这样播放器能正确分组而不是报错。row_factory = sqlite3.Row让查询结果可以按列名取值，比fetchone()[0]可读性强得多。

6.2 验证方法：先核对统计数字，再抽查播放

导出完成后不要直接丢给播放器。先做一次前端核对：用 SQL 查出来的分类数量，和 M3U 里group-title的实际分类数量要一致。一个小脚本就能对账：

# 数据库里有哪些分类 sqlite3 iptv.db "SELECT group_name, COUNT(*) FROM channel GROUP BY group_name;" # 导出文件里有哪些分类 grep -o 'group-title="[^"]*"' clean.m3u | sort | uniq -c

两个结果差在哪，哪就有清洗遗漏。然后随机挑三个分类各抽两条源，放到播放器里确认能出画面、声音正常。这一步是最终验收，测通了再决定是否大规模替换旧的播放列表。

我最初做这整套流程时，是把所有源塞进一个大文本文件，靠播放器报错反查问题源，每天修修补补。后来改成"文本拆包 → 入库 → 检测 → 导出"的流程，每接一批新数据先跑一遍，再到播放器里验证。现在维护几百个频道，稳定多了——至少不再每天靠人眼去翻失效源。希望帮到你。

本文还有配套的精品资源，点击获取