RUN.sh
· 394 B · Bash
Bruto
# 保存脚本为 compare_manifests.py,并赋可执行权限
chmod +x compare_manifests.py
# 运行脚本,依次传入 Ubuntu、Zorin、Anduin 三个 manifest 文件路径
./compare_manifests.py \
"/media/anduin/Ubuntu 25.04 amd64/casper/filesystem.manifest" \
"/media/anduin/Zorin OS 17.3 Core 64bit/casper/filesystem.manifest" \
"/media/anduin/anduinos/casper/filesystem.manifest"
1 | # 保存脚本为 compare_manifests.py,并赋可执行权限 |
2 | chmod +x compare_manifests.py |
3 | |
4 | # 运行脚本,依次传入 Ubuntu、Zorin、Anduin 三个 manifest 文件路径 |
5 | ./compare_manifests.py \ |
6 | "/media/anduin/Ubuntu 25.04 amd64/casper/filesystem.manifest" \ |
7 | "/media/anduin/Zorin OS 17.3 Core 64bit/casper/filesystem.manifest" \ |
8 | "/media/anduin/anduinos/casper/filesystem.manifest" |
9 |
compare_manifest.sh
· 5.1 KiB · Bash
Bruto
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
compare_manifests.py
读取三个 filesystem.manifest 文件,比较包的并集,并输出包含 Section、Priority、Description 及各发行版存在情况的 Markdown 表格。
增加详细日志输出,并使用多线程加速 apt show 调用。
修复:自动剥离包名中的 ":架构" 后缀。
"""
import sys
import subprocess
import logging
import concurrent.futures
import time
# 日志配置
logging.basicConfig(
level=logging.INFO,
format="%(asctime)s [%(levelname)s] %(message)s",
datefmt="%Y-%m-%d %H:%M:%S"
)
def parse_manifest(path):
"""
读取 manifest 文件,返回其中所有包名的集合。
自动剥离包名中的 ":<arch>" 后缀。
"""
pkgs = set()
logging.info(f"Parsing manifest: {path}")
try:
with open(path, encoding='utf-8') as f:
for line in f:
line = line.strip()
if not line or line.startswith('#'):
continue
parts = line.split()
raw_pkg = parts[0]
pkg = raw_pkg.split(':', 1)[0]
pkgs.add(pkg)
logging.info(f"Found {len(pkgs)} unique packages in {path}")
except Exception as e:
logging.error(f"Error reading manifest {path}: {e}")
sys.exit(1)
return pkgs
def get_pkg_info(pkg):
"""
调用 `apt show pkg`,解析 Section、Priority、Description(首段),
返回 dict 包含信息。
若调用失败,返回空字段。
"""
logging.debug(f"Fetching apt info for package: {pkg}")
try:
p = subprocess.run(
["apt", "show", pkg],
stdout=subprocess.PIPE,
stderr=subprocess.DEVNULL,
text=True,
encoding='utf-8',
timeout=30
)
except subprocess.TimeoutExpired:
logging.warning(f"Timeout fetching info for {pkg}")
return {"Section": "", "Priority": "", "Description": ""}
if p.returncode != 0:
logging.debug(f"apt show failed for {pkg}")
return {"Section": "", "Priority": "", "Description": ""}
section = ""
priority = ""
desc_lines = []
in_desc = False
for line in p.stdout.splitlines():
if in_desc:
if not line.strip():
break
desc_lines.append(line.strip())
else:
if line.startswith("Section:"):
section = line.split(":", 1)[1].strip()
elif line.startswith("Priority:"):
priority = line.split(":", 1)[1].strip()
elif line.startswith("Description:"):
desc = line.split(":", 1)[1].strip()
desc_lines.append(desc)
in_desc = True
full_desc = " ".join(desc_lines)
logging.debug(f"Obtained info for {pkg}: section={section}, priority={priority}")
return {"Section": section, "Priority": priority, "Description": full_desc}
def main():
if len(sys.argv) != 4:
print(f"Usage: {sys.argv[0]} <ubuntu_manifest> <zorin_manifest> <anduin_manifest>")
sys.exit(1)
start_time = time.time()
paths = {
"Ubuntu": sys.argv[1],
"Zorin": sys.argv[2],
"Anduin": sys.argv[3],
}
# 解析 manifests
sets = {name: parse_manifest(path) for name, path in paths.items()}
all_pkgs = sorted(set.union(*sets.values()))
total = len(all_pkgs)
logging.info(f"Total unique packages to process: {total}")
# 多线程获取 apt 信息
pkg_infos = {}
max_workers = min(10, total)
logging.info(f"Starting ThreadPoolExecutor with {max_workers} workers")
with concurrent.futures.ThreadPoolExecutor(max_workers=max_workers) as executor:
future_to_pkg = {executor.submit(get_pkg_info, pkg): pkg for pkg in all_pkgs}
for idx, future in enumerate(concurrent.futures.as_completed(future_to_pkg), 1):
pkg = future_to_pkg[future]
try:
pkg_infos[pkg] = future.result()
logging.info(f"[{idx}/{total}] Processed {pkg}")
except Exception as e:
logging.error(f"Error processing {pkg}: {e}")
pkg_infos[pkg] = {"Section": "", "Priority": "", "Description": ""}
# 写入 Markdown
out_path = "comp_result.md"
logging.info(f"Writing results to {out_path}")
with open(out_path, "w", encoding="utf-8") as out:
out.write("| Package | Section | Priority | Description | Ubuntu | Zorin | Anduin |\n")
out.write("|---------|---------|----------|-------------|--------|-------|--------|\n")
for pkg in all_pkgs:
info = pkg_infos.get(pkg, {"Section": "", "Priority": "", "Description": ""})
mark = lambda name: "√" if pkg in sets[name] else ""
desc = info["Description"].replace("|", "\\|")
out.write(
f"| {pkg} | {info['Section']} | {info['Priority']} | {desc} |"
f" {mark('Ubuntu')} | {mark('Zorin')} | {mark('Anduin')} |\n"
)
elapsed = time.time() - start_time
logging.info(f"Completed in {elapsed:.2f}s")
if __name__ == "__main__":
main()
1 | #!/usr/bin/env python3 |
2 | # -*- coding: utf-8 -*- |
3 | """ |
4 | compare_manifests.py |
5 | |
6 | 读取三个 filesystem.manifest 文件,比较包的并集,并输出包含 Section、Priority、Description 及各发行版存在情况的 Markdown 表格。 |
7 | 增加详细日志输出,并使用多线程加速 apt show 调用。 |
8 | 修复:自动剥离包名中的 ":架构" 后缀。 |
9 | """ |
10 | import sys |
11 | import subprocess |
12 | import logging |
13 | import concurrent.futures |
14 | import time |
15 | |
16 | # 日志配置 |
17 | logging.basicConfig( |
18 | level=logging.INFO, |
19 | format="%(asctime)s [%(levelname)s] %(message)s", |
20 | datefmt="%Y-%m-%d %H:%M:%S" |
21 | ) |
22 | |
23 | |
24 | def parse_manifest(path): |
25 | """ |
26 | 读取 manifest 文件,返回其中所有包名的集合。 |
27 | 自动剥离包名中的 ":<arch>" 后缀。 |
28 | """ |
29 | pkgs = set() |
30 | logging.info(f"Parsing manifest: {path}") |
31 | try: |
32 | with open(path, encoding='utf-8') as f: |
33 | for line in f: |
34 | line = line.strip() |
35 | if not line or line.startswith('#'): |
36 | continue |
37 | parts = line.split() |
38 | raw_pkg = parts[0] |
39 | pkg = raw_pkg.split(':', 1)[0] |
40 | pkgs.add(pkg) |
41 | logging.info(f"Found {len(pkgs)} unique packages in {path}") |
42 | except Exception as e: |
43 | logging.error(f"Error reading manifest {path}: {e}") |
44 | sys.exit(1) |
45 | return pkgs |
46 | |
47 | |
48 | def get_pkg_info(pkg): |
49 | """ |
50 | 调用 `apt show pkg`,解析 Section、Priority、Description(首段), |
51 | 返回 dict 包含信息。 |
52 | 若调用失败,返回空字段。 |
53 | """ |
54 | logging.debug(f"Fetching apt info for package: {pkg}") |
55 | try: |
56 | p = subprocess.run( |
57 | ["apt", "show", pkg], |
58 | stdout=subprocess.PIPE, |
59 | stderr=subprocess.DEVNULL, |
60 | text=True, |
61 | encoding='utf-8', |
62 | timeout=30 |
63 | ) |
64 | except subprocess.TimeoutExpired: |
65 | logging.warning(f"Timeout fetching info for {pkg}") |
66 | return {"Section": "", "Priority": "", "Description": ""} |
67 | |
68 | if p.returncode != 0: |
69 | logging.debug(f"apt show failed for {pkg}") |
70 | return {"Section": "", "Priority": "", "Description": ""} |
71 | |
72 | section = "" |
73 | priority = "" |
74 | desc_lines = [] |
75 | in_desc = False |
76 | |
77 | for line in p.stdout.splitlines(): |
78 | if in_desc: |
79 | if not line.strip(): |
80 | break |
81 | desc_lines.append(line.strip()) |
82 | else: |
83 | if line.startswith("Section:"): |
84 | section = line.split(":", 1)[1].strip() |
85 | elif line.startswith("Priority:"): |
86 | priority = line.split(":", 1)[1].strip() |
87 | elif line.startswith("Description:"): |
88 | desc = line.split(":", 1)[1].strip() |
89 | desc_lines.append(desc) |
90 | in_desc = True |
91 | |
92 | full_desc = " ".join(desc_lines) |
93 | logging.debug(f"Obtained info for {pkg}: section={section}, priority={priority}") |
94 | return {"Section": section, "Priority": priority, "Description": full_desc} |
95 | |
96 | |
97 | def main(): |
98 | if len(sys.argv) != 4: |
99 | print(f"Usage: {sys.argv[0]} <ubuntu_manifest> <zorin_manifest> <anduin_manifest>") |
100 | sys.exit(1) |
101 | |
102 | start_time = time.time() |
103 | paths = { |
104 | "Ubuntu": sys.argv[1], |
105 | "Zorin": sys.argv[2], |
106 | "Anduin": sys.argv[3], |
107 | } |
108 | |
109 | # 解析 manifests |
110 | sets = {name: parse_manifest(path) for name, path in paths.items()} |
111 | all_pkgs = sorted(set.union(*sets.values())) |
112 | total = len(all_pkgs) |
113 | logging.info(f"Total unique packages to process: {total}") |
114 | |
115 | # 多线程获取 apt 信息 |
116 | pkg_infos = {} |
117 | max_workers = min(10, total) |
118 | logging.info(f"Starting ThreadPoolExecutor with {max_workers} workers") |
119 | |
120 | with concurrent.futures.ThreadPoolExecutor(max_workers=max_workers) as executor: |
121 | future_to_pkg = {executor.submit(get_pkg_info, pkg): pkg for pkg in all_pkgs} |
122 | for idx, future in enumerate(concurrent.futures.as_completed(future_to_pkg), 1): |
123 | pkg = future_to_pkg[future] |
124 | try: |
125 | pkg_infos[pkg] = future.result() |
126 | logging.info(f"[{idx}/{total}] Processed {pkg}") |
127 | except Exception as e: |
128 | logging.error(f"Error processing {pkg}: {e}") |
129 | pkg_infos[pkg] = {"Section": "", "Priority": "", "Description": ""} |
130 | |
131 | # 写入 Markdown |
132 | out_path = "comp_result.md" |
133 | logging.info(f"Writing results to {out_path}") |
134 | with open(out_path, "w", encoding="utf-8") as out: |
135 | out.write("| Package | Section | Priority | Description | Ubuntu | Zorin | Anduin |\n") |
136 | out.write("|---------|---------|----------|-------------|--------|-------|--------|\n") |
137 | |
138 | for pkg in all_pkgs: |
139 | info = pkg_infos.get(pkg, {"Section": "", "Priority": "", "Description": ""}) |
140 | mark = lambda name: "√" if pkg in sets[name] else "" |
141 | desc = info["Description"].replace("|", "\\|") |
142 | out.write( |
143 | f"| {pkg} | {info['Section']} | {info['Priority']} | {desc} |" |
144 | f" {mark('Ubuntu')} | {mark('Zorin')} | {mark('Anduin')} |\n" |
145 | ) |
146 | |
147 | elapsed = time.time() - start_time |
148 | logging.info(f"Completed in {elapsed:.2f}s") |
149 | |
150 | if __name__ == "__main__": |
151 | main() |
152 |