网络蝙蝠侠部落

发表于 更新于 分类于

收视设备信息

  • 光猫型号: HG6142A
  • 盒子型号: TR100 IPTV-G9
  • 连接方式: LAN2
  • 单播组播：单播 & 组播
  • 认证时效: 7天(省内跨市联通网络可看)
  • 连接模式：桥接
  • VLAN ID: 45

获取IPTV频道

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
import requests
import json
import os
import sys

# 设置工作目录为脚本所在目录
os.chdir(os.path.dirname(os.path.abspath(__file__)))

s = requests.Session()
s.headers['User-Agent'] = 'okhttp/3.10.0'

# 获取所有频道信息
all_channels_response = s.get("http://120.87.12.38:8083/epg/api/custom/getAllChannel.json")
all_channels_data = all_channels_response.json()

if all_channels_data['status'] != '200' or 'channels' not in all_channels_data:
    print("错误：无法获取频道列表或频道列表格式错误。")
    sys.exit(1)

# 提取频道代码
channel_codes = [channel['params']['hwcode'] for channel in all_channels_data['channels'] if 'params' in channel and 'hwcode' in channel['params']]
channel_codes_str = ','.join(channel_codes)

# 获取访问令牌
token_response = s.get("http://120.87.11.11:33200/EPG/oauth/v2/token?grant_type=EncryToken&client_id=shuma2&authinfo=52A57E9F6583C26C2F2E0C6E3FC7452F164A4C84BDEE623FF387E69D25C4C1A57031D44BBAA6DEF42155B7524C4DED856EE61F320E7CFEDEA491B8AFDF90FF4B4830CEE38BAAEC1E88165B70BC52666557400555BED665177880021B8C65ECBF&UserID=0175205891367&DeviceType=TR100-G9&DeviceVersion=219.470.107&userdomain=1&datadomain=1&accountType=1")
access_token = token_response.json().get('access_token')

if not access_token:
    print("错误：未能获取有效的 access token。")
    sys.exit(1)

# 使用访问令牌请求数据
headers = {
    'Authorization': access_token,
    'User-Agent': 'okhttp/3.10.0',
    'Connection': 'Keep-Alive',
    'Content-Type': 'application/json;charset=utf-8'
}
data = json.dumps({"channelcodes": channel_codes_str})
# 请修改此处IP，确保与鉴权URL的IP一致
response = s.post("http://120.87.11.11:33200/EPG/interEpg/channellist/batch", headers=headers, data=data)

if not response.json():
    print("错误：返回的数据为空。")
    sys.exit(1)

# 保存响应数据到文件
with open('iptv.json', 'w', encoding='UTF-8') as f:
    json.dump(response.json(), f)

# 复制 base.txt 到 iptv.txt 和 iptv.m3u8
with open('base.txt', 'r', encoding='UTF-8') as f_base, open('iptv.txt', 'w', encoding='UTF-8') as f_iptv, open('iptv.m3u8', 'w', encoding='UTF-8') as f_m3u8:
    base_content = f_base.read()
    f_iptv.write(base_content)
    f_m3u8.write("#EXTM3U\n")

# 读取 iptv.json 并处理数据
with open('iptv.json', 'r', encoding='UTF-8') as f:
    data = json.load(f)
    if not data.get('channellist'):
        print("错误：'channellist' 数据为空。")
        sys.exit(1)

    channels = {}

    for item in data['channellist']:
        hwcode = item['channelcode']
        channel_info = next((ch for ch in all_channels_data['channels'] if ch['params'].get('hwcode') == hwcode), None)
        if not channel_info:
            continue 

        channel_name = channel_info['title']
        timeshift_url = item['timeshifturl']
        group = "其它"
        if "CCTV" in channel_name or "CGTN" in channel_name:
            group = "央视"
        elif any(x in channel_name for x in ["广东", "广州", "大湾区", "南方", "岭南"]):
            group = "广东"
        elif "卫视" in channel_name:
            group = "卫视"

        if group:
            if group not in channels:
                channels[group] = []
            channels[group].append((channel_name, timeshift_url))

    # 写入处理后的数据到 iptv.txt 和 iptv.m3u8
    with open('iptv.txt', 'a', encoding='UTF-8') as f_iptv, open('iptv.m3u8', 'a', encoding='UTF-8') as f_m3u8:
        for group, ch_list in channels.items():
            f_iptv.write(f"{group},#genre#\n")
            for channel_name, url in ch_list:
                f_iptv.write(f"{channel_name},{url}\n")
                f_m3u8.write(f"#EXTINF:-1,{channel_name}\n{url}\n")

发表于 更新于 分类于

事以至此

想必以下方法你们都搞过啦

  • 修改注册表；
  • 修改组策略；
  • 添加凭证项；
  • 替换动态库；
  • 关闭防火墙。

不妨试试

  • 卸载共享主机打印驱动；
  • 重新安装共享主机驱动；
  • 重新连接共享主机打印；
  • 直至客机提示安装驱动。

发表于 更新于 分类于

设备信息

  • ASUS RT-AC 1200GU （F/W Ver:V3.0.0.4.380_10732 H/W Ver:A)
  • openwrt-23.05.4-ramips-mt7621-asus_rt-ac57u-v1-initramfs-kernel.bin
  • openwrt-23.05.4-ramips-mt7621-asus_rt-ac57u-v1-squashfs-sysupgrade.bin

改国内源

vi /etc/opkg/distfeeds.conf

1
2
3
4
5
6
src/gz openwrt_core https://mirrors.aliyun.com/openwrt/releases/23.05.4/targets/x86/64/packages
src/gz openwrt_base https://mirrors.aliyun.com/openwrt/releases/23.05.4/packages/x86_64/base
src/gz openwrt_luci https://mirrors.aliyun.com/openwrt/releases/23.05.4/packages/x86_64/luci
src/gz openwrt_packages https://mirrors.aliyun.com/openwrt/releases/23.05.4/packages/x86_64/packages
src/gz openwrt_routing https://mirrors.aliyun.com/openwrt/releases/23.05.4/packages/x86_64/routing
src/gz openwrt_telephony https://mirrors.aliyun.com/openwrt/releases/23.05.4/packages/x86_64/telephony

环境安装

1
2
3
4
5
6
7
8
9
root@Openwrt:# opkg install python3-base

root@Openwrt:~# opkg install python3-pip

root@Openwrt:~# /usr/bin/python3 -m pip install -i https://pypi.tuna.tsinghua.edu.cn/simple  --upgrade pip 

root@Openwrt:~# opkg install requests

root@Openwrt:~# opkg install bs4

pip清华源

1
2
3
4
5
pip install pip -U

pip config set global.index-url https://pypi.tuna.tsinghua.edu.cn/simple

pip install -i https://pypi.tuna.tsinghua.edu.cn/simple pip -U

发表于 更新于 分类于

设备信息

  • ASUS RT-AC 1200GU （F/W Ver:V3.0.0.4.380_10732 H/W Ver:A)
  • openwrt-23.05.4-ramips-mt7621-asus_rt-ac57u-v1-initramfs-kernel.bin
  • openwrt-23.05.4-ramips-mt7621-asus_rt-ac57u-v1-squashfs-sysupgrade.bin

安装插件

1
2
3
opkg update

opkg install fdisk swap-utils kmod-usb-storage kmod-fs-ext4 e2fsprogs kmod-usb-ohci kmod-usb-uhci  block-mount

磁盘分区

1
2
3
4
5
6
7
8
9
10
11
12
13
fdisk -l (查看辨别U盘外设磁盘名称，如/dev/sda。)

fdisk /dev/sda (U盘分区)

	p (查看U盘分区)
	d (删除U盘分区)
	n (新增U盘分区)
	p (选主分区类型)
	1 (分区数量选壹)
	Enter(按回车键)
	Enter(按回车键)
	w (保存分区配置)
	reboot(重启启动)

格式分区

1
2
3
fdisk -l (查看分区名称，如/dev/sda1。)

mkfs.ext4 /dev/sda (格式化分区为ext4)

挂载分区

1
2
3
mkdir /mnt/sda1

mount /dev/sda1 /mnt/sda1

数据迁移

1
2
3
cp -r /overlay/* /mnt/sda1

block info (查看U盘的UUID)

配置挂载

1
登录路由器管理界面 - 系统 - 挂载点 - 添加 - UUID、挂载点（作为外部overlay使用） - 保存应用

完成

发表于 更新于 分类于

ASN 和 BGP 的关系

  • ASN（自治系统号）：是用于标识一个自治系统（AS）的唯一编号。一个 AS 是由一个或多个网络组成的，它们在技术和管理上由同一个实体控制。
  • BGP（边界网关协议）：是用来交换自治系统之间路由信息的协议。如果一个 ASN 和其他 ASN 之间存在对等关系（PEER），它通常是通过 BGP 来实现的。

当一个 IP 地址与某个 ASN 关联时，这意味着这个 ASN 宣告了该 IP 地址的路由信息。如果该 ASN 与多个其他 ASN 存在对等互联关系（PEER 或 TRANSIT），就表明这个 IP 地址可能是 BGP 路由的一部分。

判断是否 BGP 路由

如果工具（如 HE.NET 的 BGP 查询工具）显示：

  • 该 IP 对应某个 ASN；
  • 该 ASN 和其他多个 ASN（如 B 和 C）有 PEER 或 TRANSIT 关系；
  • 这些对等关系是 BGP 特有的连接类型，

那么可以判断这个 IP 是通过 BGP 宣告的线路。

以下是关键术语的意义：

  • PEER（对等关系）：两个 ASN 之间互相交换路由信息，但通常只交换彼此的客户流量，而不会传递第三方流量。
  • TRANSIT（中继关系）：一个 ASN 向另一个 ASN 提供更广泛的互联网接入（包括其所有的下游路由）。
  • UPSTREAM（上游关系）：你的 ASN 通过其他 ASN 接入互联网。

判断依据

假设在查询中，得到了以下信息：

  • IP 地址属于 ASN 65001（运营商 A）。
  • ASN 65001 和以下 ASN 存在 PEER 关系：
    • ASN 65002（运营商 B）
    • ASN 65003（运营商 C）

可以推断：

  1. ASN 65001 宣告了该 IP 地址的路由信息。
  2. PEER 关系意味着 ASN 65001 和其他 ASN（如 65002、65003）通过 BGP 互通。
  3. 因此，这个 IP 地址是通过 BGP 路由的线路。

补充说明

如果需要更精确的判断，可以进一步验证：

  1. 检查路由表或 Traceroute： 使用 Traceroute 查看数据包是否经过多个 ASN 的边界跳跃。例如，Traceroute 输出中会显示 ASN 的 IP 段变化，通常表示 BGP 路由在工作。
  2. ASN 的大小和互联关系： 如果一个 ASN 具有多个 PEER 或 TRANSIT 关系，且它是全球或区域范围的大型运营商（如中国电信、联通、移动等），它很可能是通过 BGP 提供服务。
  3. 询问运营商： 联系运营商，确认他们是否提供 BGP 服务，以及是否支持多运营商互联。

特殊情况

如果一个 ASN 仅与一个上游 ASN 互联，并且没有其他对等关系，那么它可能只是一个单运营商的专有线路，不能算作典型的 BGP 多线。

结论

如果 IP 对应的 ASN 与多个其他 ASN 存在 PEER 或 TRANSIT 关系，可以判断该线路是 BGP 的线路。

发表于 更新于 分类于

收视设备信息

  • 光猫型号: ZN-GE100
  • 盒子型号: HG680-KA
  • 连接方式: LAN or iTV
  • 单播组播：单播
  • 认证时效: 一个月

获取全部频道

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
import requests
from bs4 import BeautifulSoup
import re

# 1. 身份验证信息
auth_data = {
    "UserID": "0209170330290",
    "Lang": "",
    "SupportHD": "1",
    "NetUserID": "0209170330290@iptv.gd",
    "DHCPUserID": "0209170330290@iptv.gd",
    "Authenticator": "7D21702B6FA45DC20A4352BD1OOF9B694FBEFAAADD34B25FC818723AE666F7D0E6EE9B75AE7692A063041A58AC50DE673F03E178A576C195890F73D4F3C39A302FD7184CBDC91C0A0CE2AD077DA48E7A48BEC5F0BE97A15CDCAB45FF8725983F1F746501AF777AE74CB87D905E7E9E910C1A1E12FB1316689E5DD49D871EF76F",
    "STBType": "FIBERHOME_HG680KA",
    "STBVersion": "1.150150.13.2022v1",
    "conntype": "2",
    "STBID": "001005990016208092E98820115D2042",
    "templateName": "iptvsnmv3",
    "areaId": "86020",
    "userToken": "32F1583A459D7E48A6310D54FBC1CCCA",
    "userGroupId": "",
    "productPackageId": "",
    "mac": "58:19:2D:24:28:02",
    "UserField": "",
    "SoftwareVersion": "1.150150.13.2022v1",
    "IsSmartStb": "1",
    "desktopId": "",
    "stbmaker": "",
    "VIP": "",
}

# 2. 用于身份验证的URL
auth_url = "http://183.59.156.43:8082/EPG/jsp/ValidAuthenticationHWCTC.jsp"
channel_list_url = "http://183.59.156.43:8082/EPG/jsp/getchannellistHWCTC.jsp"

# 3. 创建会话
session = requests.Session()

# 4. 发起身份验证请求，获取Cookie
response = session.post(auth_url, data=auth_data)
if response.status_code == 200:
    print("身份验证成功")
else:
    print("身份验证失败")
    exit(1)

# 5. 获取频道列表页面
response = session.get(channel_list_url)
if response.status_code == 200:
    print("成功获取频道列表页面")
else:
    print("获取频道列表页面失败")
    exit(1)

# 6. 解析HTML并提取频道信息
html_content = response.text
soup = BeautifulSoup(html_content, "html.parser")
scripts = soup.find_all("script")

# 查找所有包含频道信息的脚本
m3u8_channels = ""
regex = re.compile(r',?(.+?)="(.*?)"')

for script in scripts:
    if "CTCSetConfig" in script.string:
        matches = regex.finditer(script.string)
        info = {}
        for match in matches:
            key, value = match.groups()
            if key == "ChannelName":
                m3u8_channels += f'#EXTINF:-1 tvg-name="{value}",{value}\n'
            if key == "ChannelSDP":
                m3u8_channels += f"{value}\n"

# 7. 保存m3u8文件
with open("IPTV.m3u8", "w", encoding="utf-8") as m3u8_file:
    m3u8_file.write(m3u8_channels)

print("IPTV.m3u8 文件已生成")

导出珠江频道

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
import requests
from bs4 import BeautifulSoup
import re

# 1. 身份验证信息
auth_data = {
    "UserID": "0209170330290",
    "Lang": "",
    "SupportHD": "1",
    "NetUserID": "0209170330290@iptv.gd",
    "DHCPUserID": "0209170330290@iptv.gd",
    "Authenticator": "7D21702B6FA45DC20A4352BD1OOF9B694FBEFAAADD34B25FC818723AE666F7D0E6EE9B75AE7692A063041A58AC50DE673F03E178A576C195890F73D4F3C39A302FD7184CBDC91C0A0CE2AD077DA48E7A48BEC5F0BE97A15CDCAB45FF8725983F1F746501AF777AE74CB87D905E7E9E910C1A1E12FB1316689E5DD49D871EF76F",
    "STBType": "FIBERHOME_HG680KA",
    "STBVersion": "1.150150.13.2022v1",
    "conntype": "2",
    "STBID": "001005990016208092E98820115D2042",
    "templateName": "iptvsnmv3",
    "areaId": "86020",
    "userToken": "32F1583A459D7E48A6310D54FBC1CCCA",
    "userGroupId": "",
    "productPackageId": "",
    "mac": "58:19:2D:24:28:02",
    "UserField": "",
    "SoftwareVersion": "1.150150.13.2022v1",
    "IsSmartStb": "1",
    "desktopId": "",
    "stbmaker": "",
    "VIP": "",
}

# 2. 用于身份验证的URL
auth_url = "http://183.59.156.43:8082/EPG/jsp/ValidAuthenticationHWCTC.jsp"
channel_list_url = "http://183.59.156.43:8082/EPG/jsp/getchannellistHWCTC.jsp"

# 3. 创建会话
session = requests.Session()

# 4. 发起身份验证请求，获取Cookie
response = session.post(auth_url, data=auth_data)
if response.status_code == 200:
    print("身份验证成功")
else:
    print("身份验证失败")
    exit(1)

# 5. 获取频道列表页面
response = session.get(channel_list_url)
if response.status_code == 200:
    print("成功获取频道列表页面")
else:
    print("获取频道列表页面失败")
    exit(1)

# 6. 解析HTML并提取频道信息
html_content = response.text
soup = BeautifulSoup(html_content, "html.parser")
scripts = soup.find_all("script")

# 查找所有包含频道信息的脚本
m3u8_channels = ""
regex = re.compile(r',?(.+?)="(.*?)"')
zhujiang_channel = ""

for script in scripts:
    if "CTCSetConfig" in script.string:
        matches = regex.finditer(script.string)
        info = {}
        channel_name = ""
        channel_sdp = ""
        for match in matches:
            key, value = match.groups()
            if key == "ChannelName":
                channel_name = value
            if key == "ChannelSDP":
                channel_sdp = value

        if channel_name:
            m3u8_channels += f'#EXTINF:-1 tvg-name="{channel_name}",{channel_name}\n'
            m3u8_channels += f"{channel_sdp}\n"

            # 筛选出“广东珠江”频道
            if channel_name == "广东珠江":
                zhujiang_channel = f'#EXTINF:-1 tvg-name="{channel_name}",{channel_name}\n{channel_sdp}\n'

# 7. 保存 IPTV.m3u8 文件
with open("IPTV.m3u8", "w", encoding="utf-8") as m3u8_file:
    m3u8_file.write(m3u8_channels)
print("IPTV.m3u8 文件已生成")

# 8. 保存 zhujiang.m3u8 文件
if zhujiang_channel:
    with open("zhujiang.m3u8", "w", encoding="utf-8") as zhujiang_file:
        zhujiang_file.write(zhujiang_channel)
    print("zhujiang.m3u8 文件已生成")
else:
    print("未找到“广东珠江”频道信息")

发表于 更新于 分类于

电压

  • 设备电压工作原则：给多少用多少。

电流

  • 设备电压工作原则：用多少拿多少。

电平

  • 指两功率或电压之比的对数，有时也可用来表示两电流之比的对数。常用的电平有功率电平和电压电平两类，它们各自又可分为绝对电平和相对电平两种。

  • 电平主要用于驱动信号传输，它的电流非常小，我们把高于0.7V的叫高电平，低于0.7V的叫低电平。

  • 标准低平为:0V。标准高电平为:3.3V或1.8V

电阻

串联

  • 通过同一电流；
  • 接地形成分压；
  • 串联阻值变大。

并联

  • 承受同一电压；
  • 并联形成分流；
  • 并联阻值变小。

上拉电阻

  • 在电路中一端接电压另一端接信号的电阻叫上拉电阻。（阻值较大）
  • 作用：上拉电阻可以提高电路的驱动能力，在默认状态下，将不确定的信号钳位在高电平。当输入为低时，上拉电阻又可以将信号钳位在低电平，它可以让信号在高低电平之间转换。

下拉电阻

  • 在电路中一端接地另一端接入芯片的电阻叫下拉电阻。
  • 作用：在默认状态，下拉电阻可以将不确定的信号钳位在低电平，当输入为高时，下拉电阻以可以将信号钳位在高电平，它可以让信号在高低电平之间转换。

藕合电阻

  • 在电路中芯片与芯片或芯片与接口之间传递信号经过的电阻叫耦合电阻。
  • 作用：信号匹配或缓冲。
  • 阻值一般在100欧以内（常用33Ω、47Ω、15Ω）

保险电阻

  • 在电路中两端都接电压并且阻值较小（10欧以内）的电阻叫保险电阻。
  • 作用：当输入电压过高或电流过大时，电阻熔断从而保护芯片。

检流电阻

  • 串联在电压输出端，检测电压输出端的电流。

发表于 更新于 分类于

开始准备

目标文件的局域网地址：确保文件可以通过 HTTP 或 FTP 访问。例如，http://192.168.1.100/zhujiang.m3u8。

iOS 设备与文件所在设备同处一个局域网。

快捷指令应用（iOS 自带）已安装并设置好权限。

快捷指令设置

  1. 创建快捷指令

打开 iOS 的快捷指令应用。
点击右上角的 “+” 按钮，新建快捷指令。

  1. 添加动作

  2. 获取文件内容

    搜索并添加 “获取URL的内容” 动作。
    在 URL 字段输入目标文件的局域网地址，例如 http://192.168.1.100/zhujiang.m3u8。

  3. 保存到文件

    添加 “保存文件” 动作。
    在“存储到”中选择【文件】应用中的目标文件夹。
    如果需要覆盖旧文件，启用“如果文件存在，则替换”选项。

  4. 测试运行

点击右上角 “运行” 按钮，确保文件可以成功下载并保存到指定位置。

自动定时设置

设置方法：

  1. 创建快捷指令：
    按照前述步骤，配置完成一个可用的快捷指令用于下载和保存文件。
    保存快捷指令，命名为如“更新局域网文件”。
  2. 添加日历或提醒任务：
    打开 iOS 的【提醒事项】或【日历】应用。
    创建一个重复的提醒或事件：
    设置提醒时间（例如每月的 1 号上午 10:00）。
    在提醒的“附加操作”中，选择运行“更新局域网文件”快捷指令。
  3. 运行测试：
    在提醒触发时，系统会弹出通知，点击通知即可运行快捷指令。

发表于 更新于 分类于

获取代码

python3.13

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
import sys
import os
import django

# 添加项目根目录到 sys.path 中
sys.path.append(os.path.abspath(os.path.join(os.path.dirname(__file__), '../../')))

# 设置 Django 环境变量
os.environ.setdefault("DJANGO_SETTINGS_MODULE", "epg_scraper.settings")  # 替换为你的 Django 项目名
django.setup()

import datetime
import requests
import warnings
from bs4 import BeautifulSoup as bs
from epg.models import Channel, EPG
from django.db import IntegrityError

# 定义请求头，模拟浏览器访问
headers = {
    'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/114.0.0.0 Safari/537.36'
}

# 忽略 XMLParsedAsHTMLWarning
warnings.filterwarnings('ignore')

# 获取广东电视台频道列表
def get_channels_gdtv():
    url = 'http://epg.gdtv.cn/f/1.xml'  # 获取广东电视台频道列表的XML文件
    res = requests.get(url, headers=headers)
    res.encoding = 'utf-8'
    soup = bs(res.text, 'lxml')
    contents = soup.select('channel')
    channels = []

    # 遍历频道内容并检查是否存在
    for content in contents:
        channel_id = content.attrs['id']
        name = content.ctitle.text.strip()  # 使用 strip() 清理可能的空格和换行

        # 确保频道唯一，避免重复添加
        existing_channels = Channel.objects.filter(channel_id=channel_id)
        if existing_channels.exists():
            print(f"频道 {name} 已存在于数据库。")
        else:
            Channel.objects.create(name=name, channel_id=channel_id)
            print(f"频道 {name} 已添加到数据库。")

        channels.append(existing_channels.first() if existing_channels.exists() else Channel.objects.get(channel_id=channel_id))

    return channels

# 处理节目标题中的 HTML 实体和标签
def clean_html(text):
    # 将 HTML 实体转换为普通字符（例如 & -> &）
    text = bs(text, "html.parser").get_text()

    # 替换 <br /> 标签为换行符
    text = text.replace('<br />', '\n').replace('<br>', '\n')
    
    return text.strip()

# 获取指定日期范围内的节目表
def get_epgs_gdtv(channel, start_date, end_date):
    epgs = []
    success = 1
    try:
        # 循环抓取指定日期范围内的节目
        current_date = start_date
        while current_date <= end_date:
            url = f'http://epg.gdtv.cn/f/{channel.channel_id}/{current_date.strftime("%Y-%m-%d")}.xml'
            res = requests.get(url, headers=headers, timeout=8)
            res.encoding = 'utf-8'
            soup = bs(res.text, 'lxml-xml')
            epgs_contents = soup.select('content')

            # 解析并保存节目数据
            for epga in epgs_contents:
                starttime = datetime.datetime.fromtimestamp(int(epga.attrs['time1']))
                endtime = datetime.datetime.fromtimestamp(int(epga.attrs['time2']))
                title = clean_html(epga.get_text())  # 使用清理函数

                # 检查数据库中是否已经存在相同的节目
                if not EPG.objects.filter(channel=channel, starttime=starttime, title=title).exists():
                    try:
                        # 如果不存在，则创建新的节目记录
                        EPG.objects.create(
                            channel=channel,
                            title=title,
                            starttime=starttime,
                            endtime=endtime,
                            program_date=current_date
                        )
                        print(f"新增节目: {title} ({starttime} - {endtime})")
                    except IntegrityError:
                        print(f"插入节目时发生错误: {title}")

            current_date += datetime.timedelta(days=1)  # 增加一天

        print(f"成功抓取 {channel.name} 的节目表。")
    except Exception as e:
        print(f"抓取 {channel.name} 的节目表失败: {e}")
        success = 0

    return success, epgs

# 获取前七天到后两天的节目表
def fetch_all_channels_and_epgs():
    today = datetime.datetime.now().date()
    start_date = today - datetime.timedelta(days=7)  # 七天前
    end_date = today + datetime.timedelta(days=2)  # 两天后

    channels = get_channels_gdtv()  # 获取并确保频道存在
    for channel in channels:
        get_epgs_gdtv(channel, start_date, end_date)  # 抓取并保存节目信息

# 执行抓取
if __name__ == "__main__":
    fetch_all_channels_and_epgs()

定时配置

项目settings.py添加

1
2
# Celery Beat Scheduler 设置
CELERY_BEAT_SCHEDULER = 'django_celery_beat.schedulers:DatabaseScheduler'

源码地址

1
https://github.com/heiok/epg_scraper

发表于 更新于 分类于

服务器端设置

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
[root@localhost ~]# ifconfig
eth0: flags=4163<UP,BROADCAST,RUNNING,MULTICAST>  mtu 1500
        inet 192.168.1.25  netmask 255.255.255.0  broadcast 192.168.1.255
        inet6 fe80::215:5dff:fe01:1106  prefixlen 64  scopeid 0x20<link>
        ether 00:15:5d:01:11:06  txqueuelen 1000  (Ethernet)
        RX packets 8981591  bytes 5389285998 (5.0 GiB)
        RX errors 0  dropped 58  overruns 0  frame 0
        TX packets 1369788  bytes 1311502278 (1.2 GiB)
        TX errors 0  dropped 0 overruns 0  carrier 0  collisions 0

lo: flags=73<UP,LOOPBACK,RUNNING>  mtu 65536
        inet 127.0.0.1  netmask 255.0.0.0
        inet6 ::1  prefixlen 128  scopeid 0x10<host>
        loop  txqueuelen 1000  (Local Loopback)
        RX packets 407818  bytes 117744700 (112.2 MiB)
        RX errors 0  dropped 0  overruns 0  frame 0
        TX packets 407818  bytes 117744700 (112.2 MiB)
        TX errors 0  dropped 0 overruns 0  carrier 0  collisions 0

[root@localhost ~]#
1
[root@localhost ~]# ssh -D 9527 -fgN -i id_rsa opc@180.223.1.189

代理文件配置

代理自动配置（Proxy auto-config，简称PAC）是一种网页浏览器技术，用于定义浏览器该如何自动选择适当的代理服务器来访问一个网址。

1
2
3
function FindProxyForURL(url, host) {
	return "SOCKS 192.168.1.25:9527";
}

以上三行代码保存为proxy.pac文件，放到http服务器。

1
2
3
4
5
6
7
8
DIRECT
   直接连接，不使用代理

PROXY host:port
   使用指定的http代理

SOCKS host:port
   使用指定的SOCKS代理

手机终端设置

1
设置-无线局域网-右侧感叹号-配置代理-自动-URL-http://192.168.1.203/proxy.pac-存储
0%