diff --git a/.github/workflows/custom.yml b/.github/workflows/custom.yml
index 6841a56e..7e173d04 100644
--- a/.github/workflows/custom.yml
+++ b/.github/workflows/custom.yml
@@ -138,7 +138,6 @@ jobs:
writeConfigKey "addons.\"${A}\"" "${VALUE}" "${PRESET_CONFIG_FILE}"
done
fi
- writeConfigKey "automated" "true" "${PRESET_CONFIG_FILE}"
echo "$(cat "${PRESET_CONFIG_FILE}")"
sudo echo "arc-${MODEL}-${PRODUCTVER:0:3}-${TAG}" >"/tmp/automated"
sudo mv -f "/tmp/automated" "/tmp/p3/automated"
diff --git a/scripts/functions.py b/scripts/functions.py
index 70e3bd6d..08074679 100755
--- a/scripts/functions.py
+++ b/scripts/functions.py
@@ -6,91 +6,62 @@
# See /LICENSE for more information.
#
-import os, re, sys, glob, json, yaml, click, shutil, tarfile, kmodule, requests, urllib3
-from requests.adapters import HTTPAdapter
-from requests.packages.urllib3.util.retry import Retry # type: ignore
-from openpyxl import Workbook
+import os, click
+
+WORK_PATH = os.path.abspath(os.path.dirname(__file__))
+
@click.group()
def cli():
"""
- The CLI is a commands to ARC.
+ The CLI is a commands to Arc.
"""
pass
-@cli.command()
-@click.option("-w", "--workpath", type=str, required=True, help="The workpath of ARC.")
-@click.option("-j", "--jsonpath", type=str, required=True, help="The output path of jsonfile.")
-def getmodels(workpath, jsonpath):
- models = {}
- platforms_yml = os.path.join(workpath, "mnt", "p3", "configs", "platforms.yml")
- with open(platforms_yml, "r") as f:
- P_data = yaml.safe_load(f)
- P_platforms = P_data.get("platforms", [])
- for P in P_platforms:
- productvers = {}
- for V in P_platforms[P]["productvers"]:
- kpre = P_platforms[P]["productvers"][V].get("kpre", "")
- kver = P_platforms[P]["productvers"][V].get("kver", "")
- productvers[V] = f"{kpre}-{kver}" if kpre else kver
- models[P] = {"productvers": productvers, "models": []}
+def mutually_exclusive_options(ctx, param, value):
+ other_option = "file" if param.name == "data" else "data"
+ if value is not None and ctx.params.get(other_option) is not None:
+ raise click.UsageError(f"Illegal usage: `{param.name}` is mutually exclusive with `{other_option}`.")
+ return value
- adapter = HTTPAdapter(max_retries=Retry(total=3, backoff_factor=1, status_forcelist=[500, 502, 503, 504]))
- session = requests.Session()
- session.mount("http://", adapter)
- session.mount("https://", adapter)
- urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)
-
- try:
- url = "http://update7.synology.com/autoupdate/genRSS.php?include_beta=1"
- #url = "https://update7.synology.com/autoupdate/genRSS.php?include_beta=1"
- req = session.get(url, timeout=10, verify=False)
- req.encoding = "utf-8"
- p = re.compile(r"(.*?).*?(.*?)", re.MULTILINE | re.DOTALL)
- data = p.findall(req.text)
- except Exception as e:
- click.echo(f"Error: {e}")
- return
+def validate_required_param(ctx, param, value):
+ if not value and "file" not in ctx.params and "data" not in ctx.params:
+ raise click.MissingParameter(param_decls=[param.name])
+ return value
- for item in data:
- if not "DSM" in item[1]:
- continue
- arch = item[0].split("_")[1]
- name = item[1].split("/")[-1].split("_")[1].replace("%2B", "+")
- if arch not in models:
- continue
- if name in (A for B in models for A in models[B]["models"]):
- continue
- models[arch]["models"].append(name)
-
- if jsonpath:
- with open(jsonpath, "w") as f:
- json.dump(models, f, indent=4, ensure_ascii=False)
-
-@cli.command()
-@click.option("-w", "--workpath", type=str, required=True, help="The workpath of ARC.")
-@click.option("-j", "--jsonpath", type=str, required=True, help="The output path of jsonfile.")
-def getpats(workpath, jsonpath):
- def __fullversion(ver):
- arr = ver.split('-')
- a, b, c = (arr[0].split('.') + ['0', '0', '0'])[:3]
+def __fullversion(ver):
+ out = ver
+ arr = ver.split('-')
+ if len(arr) > 0:
+ a = arr[0].split('.')[0] if len(arr[0].split('.')) > 0 else '0'
+ b = arr[0].split('.')[1] if len(arr[0].split('.')) > 1 else '0'
+ c = arr[0].split('.')[2] if len(arr[0].split('.')) > 2 else '0'
d = arr[1] if len(arr) > 1 else '00000'
e = arr[2] if len(arr) > 2 else '0'
- return f'{a}.{b}.{c}-{d}-{e}'
+ out = '{}.{}.{}-{}-{}'.format(a,b,c,d,e)
+ return out
- platforms_yml = os.path.join(workpath, "mnt", "p3", "configs", "platforms.yml")
- with open(platforms_yml, "r") as f:
- data = yaml.safe_load(f)
- platforms = data.get("platforms", [])
+@cli.command()
+@click.option("-p", "--platforms", type=str, help="The platforms of Syno.")
+def getmodels(platforms=None):
+ """
+ Get Syno Models.
+ """
+ import re, json, requests, urllib3
+ from requests.adapters import HTTPAdapter
+ from requests.packages.urllib3.util.retry import Retry # type: ignore
adapter = HTTPAdapter(max_retries=Retry(total=3, backoff_factor=1, status_forcelist=[500, 502, 503, 504]))
session = requests.Session()
session.mount("http://", adapter)
session.mount("https://", adapter)
urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)
-
+
+ PS = platforms.lower().replace(",", " ").split() if platforms else []
+
+ models = []
try:
url = "http://update7.synology.com/autoupdate/genRSS.php?include_beta=1"
#url = "https://update7.synology.com/autoupdate/genRSS.php?include_beta=1"
@@ -98,27 +69,42 @@ def getpats(workpath, jsonpath):
req = session.get(url, timeout=10, verify=False)
req.encoding = "utf-8"
p = re.compile(r"(.*?).*?(.*?)", re.MULTILINE | re.DOTALL)
- data = p.findall(req.text)
- except Exception as e:
- click.echo(f"Error: {e}")
- return
- models = []
- for item in data:
- if not "DSM" in item[1]:
- continue
- arch = item[0].split("_")[1]
- name = item[1].split("/")[-1].split("_")[1].replace("%2B", "+")
- if arch not in platforms:
- continue
- if name in models:
- continue
- models.append(name)
+ data = p.findall(req.text)
+ for item in data:
+ if not "DSM" in item[1]:
+ continue
+ arch = item[0].split("_")[1]
+ name = item[1].split("/")[-1].split("_")[1].replace("%2B", "+")
+ if PS and arch.lower() not in PS:
+ continue
+ if not any(m["name"] == name for m in models):
+ models.append({"name": name, "arch": arch})
+
+ models.sort(key=lambda k: (k["arch"], k["name"]))
+
+ except Exception as e:
+ # click.echo(f"Error: {e}")
+ pass
+
+ print(json.dumps(models, indent=4))
+
+@cli.command()
+@click.option("-m", "--model", type=str, required=True, help="The model of Syno.")
+@click.option("-v", "--version", type=str, required=True, help="The version of Syno.")
+def getpats4mv(model, version):
+ import json, requests, urllib3
+ from requests.adapters import HTTPAdapter
+ from requests.packages.urllib3.util.retry import Retry # type: ignore
+
+ adapter = HTTPAdapter(max_retries=Retry(total=3, backoff_factor=1, status_forcelist=[500, 502, 503, 504]))
+ session = requests.Session()
+ session.mount("http://", adapter)
+ session.mount("https://", adapter)
+ urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)
pats = {}
- for M in models:
- pats[M] = {}
- version = '7'
+ try:
urlInfo = "https://www.synology.com/api/support/findDownloadInfo?lang=en-us"
urlSteps = "https://www.synology.com/api/support/findUpgradeSteps?"
#urlInfo = "https://www.synology.cn/api/support/findDownloadInfo?lang=zh-cn"
@@ -126,20 +112,16 @@ def getpats(workpath, jsonpath):
major = f"&major={version.split('.')[0]}" if len(version.split('.')) > 0 else ""
minor = f"&minor={version.split('.')[1]}" if len(version.split('.')) > 1 else ""
- try:
- req = session.get(f"{urlInfo}&product={M.replace('+', '%2B')}{major}{minor}", timeout=10, verify=False)
- req.encoding = "utf-8"
- data = json.loads(req.text)
- except Exception as e:
- click.echo(f"Error: {e}")
- continue
+ req = session.get(f"{urlInfo}&product={model.replace('+', '%2B')}{major}{minor}", timeout=10, verify=False)
+ req.encoding = "utf-8"
+ data = json.loads(req.text)
build_ver = data['info']['system']['detail'][0]['items'][0]['build_ver']
build_num = data['info']['system']['detail'][0]['items'][0]['build_num']
buildnano = data['info']['system']['detail'][0]['items'][0]['nano']
V = __fullversion(f"{build_ver}-{build_num}-{buildnano}")
- if V not in pats[M]:
- pats[M][V] = {
+ if V not in pats:
+ pats[V] = {
'url': data['info']['system']['detail'][0]['items'][0]['files'][0]['url'].split('?')[0],
'sum': data['info']['system']['detail'][0]['items'][0]['files'][0]['checksum']
}
@@ -152,100 +134,120 @@ def getpats(workpath, jsonpath):
if not major or not minor:
majorTmp = f"&major={I['version'].split('.')[0]}" if len(I['version'].split('.')) > 0 else ""
minorTmp = f"&minor={I['version'].split('.')[1]}" if len(I['version'].split('.')) > 1 else ""
- try:
- reqTmp = session.get(f"{urlInfo}&product={M.replace('+', '%2B')}{majorTmp}{minorTmp}", timeout=10, verify=False)
- reqTmp.encoding = "utf-8"
- dataTmp = json.loads(reqTmp.text)
- except Exception as e:
- click.echo(f"Error: {e}")
- continue
+ reqTmp = session.get(f"{urlInfo}&product={model.replace('+', '%2B')}{majorTmp}{minorTmp}", timeout=10, verify=False)
+ reqTmp.encoding = "utf-8"
+ dataTmp = json.loads(reqTmp.text)
build_ver = dataTmp['info']['system']['detail'][0]['items'][0]['build_ver']
build_num = dataTmp['info']['system']['detail'][0]['items'][0]['build_num']
buildnano = dataTmp['info']['system']['detail'][0]['items'][0]['nano']
V = __fullversion(f"{build_ver}-{build_num}-{buildnano}")
- if V not in pats[M]:
- pats[M][V] = {
+ if V not in pats:
+ pats[V] = {
'url': dataTmp['info']['system']['detail'][0]['items'][0]['files'][0]['url'].split('?')[0],
'sum': dataTmp['info']['system']['detail'][0]['items'][0]['files'][0]['checksum']
}
for J in I['versions']:
to_ver = J['build']
- try:
- reqSteps = session.get(f"{urlSteps}&product={M.replace('+', '%2B')}&from_ver={from_ver}&to_ver={to_ver}", timeout=10, verify=False)
- if reqSteps.status_code != 200:
- continue
- reqSteps.encoding = "utf-8"
- dataSteps = json.loads(reqSteps.text)
- except Exception as e:
- click.echo(f"Error: {e}")
+ reqSteps = session.get(f"{urlSteps}&product={model.replace('+', '%2B')}&from_ver={from_ver}&to_ver={to_ver}", timeout=10, verify=False)
+ if reqSteps.status_code != 200:
continue
-
+ reqSteps.encoding = "utf-8"
+ dataSteps = json.loads(reqSteps.text)
for S in dataSteps['upgrade_steps']:
if not S.get('full_patch') or not S['build_ver'].startswith(version):
continue
V = __fullversion(f"{S['build_ver']}-{S['build_num']}-{S['nano']}")
- if V not in pats[M]:
- pats[M][V] = {
+ if V not in pats:
+ pats[V] = {
'url': S['files'][0]['url'].split('?')[0],
'sum': S['files'][0]['checksum']
}
+ except Exception as e:
+ # click.echo(f"Error: {e}")
+ pass
- if jsonpath:
- with open(jsonpath, "w") as f:
- json.dump(pats, f, indent=4, ensure_ascii=False)
-
-@cli.command()
-@click.option("-w", "--workpath", type=str, required=True, help="The workpath of ARC.")
-@click.option("-j", "--jsonpath", type=str, required=True, help="The output path of jsonfile.")
-def getaddons(workpath, jsonpath):
- AS = glob.glob(os.path.join(workpath, "mnt", "p3", "addons", "*", "manifest.yml"))
- AS.sort()
- addons = {}
- for A in AS:
- with open(A, "r") as file:
- A_data = yaml.safe_load(file)
- A_name = A_data.get("name", "")
- A_system = A_data.get("system", False)
- A_description = A_data.get("description", "")
- addons[A_name] = {"system": A_system, "description": A_description}
- if jsonpath:
- with open(jsonpath, "w") as f:
- json.dump(addons, f, indent=4, ensure_ascii=False)
+ pats = {k: pats[k] for k in sorted(pats.keys(), reverse=True)}
+ print(json.dumps(pats, indent=4))
@cli.command()
-@click.option("-w", "--workpath", type=str, required=True, help="The workpath of ARC.")
-@click.option("-j", "--jsonpath", type=str, required=True, help="The output path of jsonfile.")
-def getmodules(workpath, jsonpath):
- MS = glob.glob(os.path.join(workpath, "mnt", "p3", "modules", "*.tgz"))
- MS.sort()
- modules = {}
- TMP_PATH = "/tmp/modules"
- if os.path.exists(TMP_PATH):
- shutil.rmtree(TMP_PATH)
- for M in MS:
- M_name = os.path.splitext(os.path.basename(M))[0]
- M_modules = {}
- os.makedirs(TMP_PATH)
- with tarfile.open(M, "r") as tar:
- tar.extractall(TMP_PATH)
- KS = glob.glob(os.path.join(TMP_PATH, "*.ko"))
- KS.sort()
- for K in KS:
- K_name = os.path.splitext(os.path.basename(K))[0]
- K_info = kmodule.modinfo(K, basedir=os.path.dirname(K), kernel=None)[0]
- K_description = K_info.get("description", "")
- K_depends = K_info.get("depends", "")
- M_modules[K_name] = {"description": K_description, "depends": K_depends}
- modules[M_name] = M_modules
- if os.path.exists(TMP_PATH):
- shutil.rmtree(TMP_PATH)
- if jsonpath:
- with open(jsonpath, "w") as file:
- json.dump(modules, file, indent=4, ensure_ascii=False)
+@click.option("-p", "--models", type=str, help="The models of Syno.")
+def getpats(models=None):
+ import re, json, requests, urllib3
+ from bs4 import BeautifulSoup
+ from requests.adapters import HTTPAdapter
+ from requests.packages.urllib3.util.retry import Retry # type: ignore
+ adapter = HTTPAdapter(max_retries=Retry(total=3, backoff_factor=1, status_forcelist=[500, 502, 503, 504]))
+ session = requests.Session()
+ session.mount("http://", adapter)
+ session.mount("https://", adapter)
+ urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)
+
+ MS = models.lower().replace(",", " ").split() if models else []
+
+ pats = {}
+ try:
+ req = session.get('https://archive.synology.com/download/Os/DSM', timeout=10, verify=False)
+ req.encoding = 'utf-8'
+ bs = BeautifulSoup(req.text, 'html.parser')
+ p = re.compile(r"(.*?)-(.*?)", re.MULTILINE | re.DOTALL)
+ l = bs.find_all('a', string=p)
+ for i in l:
+ ver = i.attrs['href'].split('/')[-1]
+ if not ver.startswith('7'):
+ continue
+ req = session.get(f'https://archive.synology.com{i.attrs["href"]}', timeout=10, verify=False)
+ req.encoding = 'utf-8'
+ bs = BeautifulSoup(req.text, 'html.parser')
+ p = re.compile(r"DSM_(.*?)_(.*?).pat", re.MULTILINE | re.DOTALL)
+ data = bs.find_all('a', string=p)
+ for item in data:
+ rels = p.search(item.attrs['href'])
+ if rels:
+ model, _ = rels.groups()
+ model = model.replace('%2B', '+')
+ if MS and model.lower() not in MS:
+ continue
+ if model not in pats:
+ pats[model] = {}
+ pats[model][__fullversion(ver)] = item.attrs['href']
+ except Exception as e:
+ # click.echo(f"Error: {e}")
+ pass
+
+ print(json.dumps(pats, indent=4))
+
+@cli.command()
+@click.option("-p", "--platforms", type=str, help="The platforms of Syno.")
+def getmodelsoffline(platforms=None):
+ """
+ Get Syno Models.
+ """
+ import json
+ import os
+
+ PS = platforms.lower().replace(",", " ").split() if platforms else []
+
+ with open(os.path.join('/mnt/p3/configs', "offline.json")) as user_file:
+ data = json.load(user_file)
+
+ models = []
+ for item in data["channel"]["item"]:
+ if not item["title"].startswith("DSM"):
+ continue
+ for model in item["model"]:
+ arch = model["mUnique"].split("_")[1]
+ name = model["mLink"].split("/")[-1].split("_")[1].replace("%2B", "+")
+ if PS and arch.lower() not in PS:
+ continue
+ if not any(m["name"] == name for m in models):
+ models.append({"name": name, "arch": arch})
+
+ models = sorted(models, key=lambda k: (k["arch"], k["name"]))
+ print(json.dumps(models, indent=4))
if __name__ == "__main__":
cli()
\ No newline at end of file