moyan/pmpt
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

添加makefile

Browse Source
This commit is contained in:
root 2024-04-03 21:12:35 +08:00
parent a135ef0e1d
commit 94642c4df5
12 changed files with 461 additions and 351 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

48
Makefile Normal file
View File

@ -0,0 +1,48 @@
# Makefile for Python project
# 定义变量
PYTHON := python
PMPT := python -m pmpt
PIP := pip
VENV := venv
# 定义目录
DIST_DIR := dists
BUILD_DIR := build
# 定义目标
ALL: help
help:
@echo "Available targets:"
@echo " install Install dependencies"
@echo " test Run tests"
@echo " clean Clean up"
build:
$(PYTHON) setup.py bdist_wheel -d $(DIST_DIR) sdist -d $(DIST_DIR)
init: venv
@for file in requirements/*.txt; do \
if [ -f "$$file" ]; then \
$(PIP) install -r "$$file"; \
fi; \
done
venv:
$(PYTHON) -m venv $(VENV)
cleanall: clean
rm -r $(VENV)
clean:
rm -r $(DIST_DIR) $(BUILD_DIR) pmpt.egg-info
push:
@if [ -z "$(msg)" ]; then \
echo "Please provide a commit message using 'msg' parameter."; \
exit 1; \
else \
git add .; \
git commit -m "$(msg)"; \
git push origin master; \
fi

97
pmpt/__init__.py
View File

@ -11,102 +11,117 @@ from . import search as searchs
import datetime import datetime
import webbrowser import webbrowser
import time import time
@click.group() @click.group()
def cli(): def cli():
today = datetime.datetime.today() today = datetime.datetime.today()
if today.month == 9 and today.day == 28: if today.month == 9 and today.day == 28:
webbrowser.open('https://ys.mihoyo.com/') webbrowser.open("https://ys.mihoyo.com/")
util.console.print('[green]Genshin！activate！[/green]') util.console.print("[green]Genshin！activate！[/green]")
time.sleep(0.7) time.sleep(0.7)
util.console.print('[green]You all must download Genshin Impact.[/green]') util.console.print("[green]You all must download Genshin Impact.[/green]")
time.sleep(0.7) time.sleep(0.7)
util.console.print("Then don't use PMPT.",style='green') util.console.print("Then don't use PMPT.", style="green")
try: try:
import pip import pip
except ImportError: except ImportError:
util.logger.critical('没有pip') util.logger.critical("没有pip")
util.console.print('❌ [red]pip module not found![/red]') util.console.print("❌ [red]pip module not found![/red]")
exit(1) exit(1)
@cli.command(short_help='Update Package Index')
@cli.command(short_help="Update Package Index")
def update(): def update():
# 调用更新函数 # 调用更新函数
updates.getAllIndex() updates.getAllIndex()
@cli.command(short_help='Install Python package')
@click.argument('packlist',nargs=-1, required=True) @cli.command(short_help="Install Python package")
@click.option('--upgrade','-U',is_flag=True,default=False) @click.argument("packlist", nargs=-1, required=True)
@click.option('--reads', '-r',is_flag=True,default=False) @click.option("--upgrade", "-U", is_flag=True, default=False)
@click.option('--force-reinstall','-fr',is_flag=True,default=False) @click.option("--reads", "-r", is_flag=True, default=False)
@click.option('--ignore-requires-python','-irp',is_flag=True,default=False) @click.option("--force-reinstall", "-fr", is_flag=True, default=False)
@click.option('--yes','-y',is_flag=True,default=False) @click.option("--ignore-requires-python", "-irp", is_flag=True, default=False)
@click.option('--command','-c',is_flag=True,default=False) @click.option("--yes", "-y", is_flag=True, default=False)
@click.option("--command", "-c", is_flag=True, default=False)
def install(*args, **kwargs): def install(*args, **kwargs):
installs.main(*args, **kwargs) installs.main(*args, **kwargs)
@cli.command(name='list',short_help='List all Python packages')
@cli.command(name="list", short_help="List all Python packages")
def listp(): def listp():
table = Table(show_header=True) table = Table(show_header=True)
table.add_column('Name') table.add_column("Name")
table.add_column('Version') table.add_column("Version")
listv = util.runpip('freeze',out=False) listv = util.runpip("freeze", out=False)
for line in iter(listv.stdout.readline, b''): for line in iter(listv.stdout.readline, b""):
# 在这里你可以对每一行输出进行处理 # 在这里你可以对每一行输出进行处理
line = line.decode('utf-8').strip() # 将字节转换为字符串并去除换行符 line = line.decode("utf-8").strip() # 将字节转换为字符串并去除换行符
if '==' not in line or line[0]=='#': if "==" not in line or line[0] == "#":
continue continue
lineList = line.split('==') lineList = line.split("==")
table.add_row(lineList[0], lineList[1]) table.add_row(lineList[0], lineList[1])
util.console.print(table) util.console.print(table)
@cli.command() @cli.command()
@click.argument('name') @click.argument("name")
@click.option('--yes','-y',is_flag=True,default=False) @click.option("--yes", "-y", is_flag=True, default=False)
def remove(name, yes): def remove(name, yes):
args = [] args = []
if yes: if yes:
args.append('-y') args.append("-y")
args.append(name) args.append(name)
util.runpip('uninstall',args) util.runpip("uninstall", args)
@cli.command() @cli.command()
@click.argument('name') @click.argument("name")
@click.option('--allinfo','-a',is_flag=True,default=False) @click.option("--allinfo", "-a", is_flag=True, default=False)
@click.option('--api-url','-u',default=None) @click.option("--api-url", "-u", default=None)
def search(*args, **kwargs): def search(*args, **kwargs):
searchs.main(*args, **kwargs) searchs.main(*args, **kwargs)
@cli.group() @cli.group()
def source(): def source():
pass pass
@source.command() @source.command()
@click.argument('url') @click.argument("url")
@click.option('--priority','-p',default=1,type=int) @click.option("--priority", "-p", default=1, type=int)
def add(*args, **kwargs): def add(*args, **kwargs):
sou.add(*args, **kwargs) sou.add(*args, **kwargs)
@source.command(name='list')
@source.command(name="list")
def lists(): def lists():
sou.lists() sou.lists()
@source.command(name='remove')
@click.argument('ids',default=None,required=False) @source.command(name="remove")
@click.option('-y','--yes',is_flag=True,default=False) @click.argument("ids", default=None, required=False)
@click.option("-y", "--yes", is_flag=True, default=False)
def removes(*args, **kwargs): def removes(*args, **kwargs):
sou.remove(*args, **kwargs) sou.remove(*args, **kwargs)
@source.command(name='modify')
@click.argument('ids') @source.command(name="modify")
@click.argument('key') @click.argument("ids")
@click.argument('val') @click.argument("key")
@click.argument("val")
def modifys(*args, **kwargs): def modifys(*args, **kwargs):
sou.modify(*args, **kwargs) sou.modify(*args, **kwargs)
@cli.command() @cli.command()
def version(): def version():
environ.main() environ.main()
if __name__ == '__main__':
if __name__ == "__main__":
cli() cli()

1
pmpt/__main__.py
View File

@ -1,2 +1,3 @@
import pmpt import pmpt
pmpt.cli() pmpt.cli()

38
pmpt/environment.py
View File

@ -3,50 +3,62 @@ import re
from . import util from . import util
import moyanlib import moyanlib
def get_version(command): def get_version(command):
try: try:
# 使用 subprocess 调用系统命令获取版本信息 # 使用 subprocess 调用系统命令获取版本信息
output = subprocess.check_output(command, shell=True, stderr=subprocess.STDOUT, universal_newlines=True) output = subprocess.check_output(
command, shell=True, stderr=subprocess.STDOUT, universal_newlines=True
)
# 使用正则表达式提取版本号 # 使用正则表达式提取版本号
version_match = re.search(r'(\d+\.\d+(\.\d+){0,2})', output) version_match = re.search(r"(\d+\.\d+(\.\d+){0,2})", output)
if version_match: if version_match:
return version_match.group(1) return version_match.group(1)
else: else:
return '[red]False[/red]' return "[red]False[/red]"
except subprocess.CalledProcessError: except subprocess.CalledProcessError:
return '[red]False[/red]' return "[red]False[/red]"
def getGCCver(): def getGCCver():
# 检查是否能执行 gcc 命令 # 检查是否能执行 gcc 命令
return get_version('gcc --version') return get_version("gcc --version")
def getClangVer(): def getClangVer():
# 检查是否能执行 clang 命令 # 检查是否能执行 clang 命令
return get_version('clang --version') return get_version("clang --version")
def getMSVCver(): def getMSVCver():
# 检查是否能执行 cl 命令（MSVC编译器） # 检查是否能执行 cl 命令（MSVC编译器）
return get_version('cl') return get_version("cl")
def getRustVer(): def getRustVer():
process = subprocess.Popen(['rustc --version'], stdout=subprocess.PIPE, stderr=subprocess.PIPE,shell=True) process = subprocess.Popen(
["rustc --version"], stdout=subprocess.PIPE, stderr=subprocess.PIPE, shell=True
)
stdout, stderr = process.communicate() stdout, stderr = process.communicate()
if process.returncode == 0: if process.returncode == 0:
rustVer = stdout.decode().split(' ')[1] rustVer = stdout.decode().split(" ")[1]
return rustVer return rustVer
else: else:
return '[red]False[/red]' return "[red]False[/red]"
def getPIPver(): def getPIPver():
return get_version('pip -V') return get_version("pip -V")
def main(): def main():
info = moyanlib.getInfo() info = moyanlib.getInfo()
printText = f'''[white]PMPT {util.__version__}[/white] printText = f"""[white]PMPT {util.__version__}[/white]
{info['OS']['Name']} {info['OS']['Version']} {info['OS']['Name']} {info['OS']['Version']}
Python Version: {info['Python']['Version']} Python Version: {info['Python']['Version']}
PIP Version: [green]{getPIPver()}[/green] PIP Version: [green]{getPIPver()}[/green]
GCC Version: [green]{getGCCver()}[/green] GCC Version: [green]{getGCCver()}[/green]
Clang Version: [green]{getClangVer()}[/green] Clang Version: [green]{getClangVer()}[/green]
MSVC Version: [green]{getMSVCver()}[/green] MSVC Version: [green]{getMSVCver()}[/green]
Rust Version: [green]{getRustVer()}[/green]''' Rust Version: [green]{getRustVer()}[/green]"""
util.console.print(printText) util.console.print(printText)

75
pmpt/install.py
View File

@ -3,9 +3,9 @@ import requests
from rich.table import Table from rich.table import Table
from urllib import parse from urllib import parse
def search(name): def search(name):
util.loadIndex() #加载索引 for Index in util.loadIndex():
for Index in util.IndexList:
dt = Index.packageList.get(name, None) dt = Index.packageList.get(name, None)
if dt: if dt:
try: try:
@ -19,35 +19,35 @@ def search(name):
return Index.IndexURL return Index.IndexURL
def main(
def main(packlist,upgrade,reads,force_reinstall,ignore_requires_python,yes,command): packlist, upgrade, reads, force_reinstall, ignore_requires_python, yes, command
):
console = util.console console = util.console
if reads: # 从文件读取列表 if reads: # 从文件读取列表
f = open(packlist[0]) f = open(packlist[0])
packlist = f.read().split('\n') packlist = f.read().split("\n")
packsInfo = {} packsInfo = {}
with console.status('🚀🔍 Searching for package information...') as status: with console.status("🚀🔍 Searching for package information...") as status:
# 创建表格 # 创建表格
table = Table(show_header=True,header_style='bold') table = Table(show_header=True, header_style="bold")
table.add_column("Package Name", width=20,style='bold') table.add_column("Package Name", width=20, style="bold")
table.add_column("Package Source", width=20,style='green') table.add_column("Package Source", width=20, style="green")
for rawpack in packlist: # 解析指定了版本的包名 for rawpack in packlist: # 解析指定了版本的包名
if '==' in rawpack: if "==" in rawpack:
pack = rawpack.split('==')[0] pack = rawpack.split("==")[0]
elif '>=' in rawpack: elif ">=" in rawpack:
pack = rawpack.split('>=')[0] pack = rawpack.split(">=")[0]
elif '<=' in rawpack: elif "<=" in rawpack:
pack = rawpack.split('<=')[0] pack = rawpack.split("<=")[0]
elif '<' in rawpack: elif "<" in rawpack:
pack = rawpack.split('<')[0] pack = rawpack.split("<")[0]
elif '>' in rawpack: elif ">" in rawpack:
pack = rawpack.split('>')[0] pack = rawpack.split(">")[0]
else: else:
pack = rawpack pack = rawpack
result = search(pack.lower()) # 转小写并获取源地址 result = search(pack.lower()) # 转小写并获取源地址
packsInfo[pack] = [result, rawpack] packsInfo[pack] = [result, rawpack]
@ -61,47 +61,44 @@ def main(packlist,upgrade,reads,force_reinstall,ignore_requires_python,yes,comma
console.print(table) console.print(table)
if len(canInstallPack) < 1: if len(canInstallPack) < 1:
console.print('❌ [red]There are no packages available for installation.[red]') console.print("❌ [red]There are no packages available for installation.[red]")
exit(1) exit(1)
console.print('📦 Packages to be installed:') console.print("📦 Packages to be installed:")
console.print(' '.join(canInstallPack)) console.print(" ".join(canInstallPack))
while True: # 是否允许安装 while True: # 是否允许安装
if yes: if yes:
break break
ye = console.input('Proceed with installation? [Y/n]: ') ye = console.input("Proceed with installation? [Y/n]: ")
if ye.lower() == 'y': if ye.lower() == "y":
break break
elif ye.lower() == 'n': elif ye.lower() == "n":
console.print('🛑 [red]User canceled the installation.[/red]') console.print("🛑 [red]User canceled the installation.[/red]")
exit(1) exit(1)
else: else:
continue continue
console.print('🛠️ Initiating pip installation...') console.print("🛠️ Initiating pip installation...")
for pack in canInstallPack: for pack in canInstallPack:
# 构建命 # 构建命
args = [ '-i', packsInfo[pack][0]] # 指定源 args = ["-i", packsInfo[pack][0]] # 指定源
if upgrade: # 升级 if upgrade: # 升级
args.append('-U') args.append("-U")
if force_reinstall: # 强制重新安装 if force_reinstall: # 强制重新安装
args.append('--force-reinstall') args.append("--force-reinstall")
if ignore_requires_python: # 忽略Python版本 if ignore_requires_python: # 忽略Python版本
args.append('--ignore-requires-python') args.append("--ignore-requires-python")
args.append(packsInfo[pack][1]) args.append(packsInfo[pack][1])
with console.status('🚀 [green]Installing...[/green]') as status: with console.status("🚀 [green]Installing...[/green]") as status:
ret = util.runpip('install',args,command) # 运行pip ret = util.runpip("install", args, command) # 运行pip
if ret.returncode != 0: # 是否执行完毕 if ret.returncode != 0: # 是否执行完毕
console.print('❌ [red]Installation failed.[/red]') console.print("❌ [red]Installation failed.[/red]")
exit(1) exit(1)
console.print(f'✅ [green]Installation successful for {pack}[/green]') console.print(f"✅ [green]Installation successful for {pack}[/green]")

34
pmpt/search.py
View File

@ -5,36 +5,36 @@ from .install import search
import os import os
console = util.console console = util.console
def main(name, allinfo, api_url): def main(name, allinfo, api_url):
if not search(name): if not search(name):
console.print('❌ [red]The package does not exist[/red]') console.print("❌ [red]The package does not exist[/red]")
exit() exit()
if not api_url: if not api_url:
api_url = open(os.path.join(util.dirs.user_config_dir,'api.url')).read() api_url = open(os.path.join(util.dirs.user_config_dir, "api.url")).read()
req = requests.get(api_url.format(name)) req = requests.get(api_url.format(name))
if req.status_code == 404: if req.status_code == 404:
console.print(404) console.print(404)
console.print('❌ [red]The package does not exist[/red]') console.print("❌ [red]The package does not exist[/red]")
exit() exit()
elif req.status_code != 200: elif req.status_code != 200:
console.print('❌ [red]Server Error![/red]') console.print("❌ [red]Server Error![/red]")
console.print('[red]Error Code: '+str(req.status_code)+'[/red]') console.print("[red]Error Code: " + str(req.status_code) + "[/red]")
exit() exit()
packageInfo = jsons.loads(req.text) packageInfo = jsons.loads(req.text)
if not allinfo: if not allinfo:
print('Package Name:',packageInfo['info']['name']) print("Package Name:", packageInfo["info"]["name"])
print('Version:',packageInfo['info']['version']) print("Version:", packageInfo["info"]["version"])
print('Author:',packageInfo['info']['author']) print("Author:", packageInfo["info"]["author"])
print('Summary:',packageInfo['info']['summary']) print("Summary:", packageInfo["info"]["summary"])
print('Keywords:',packageInfo['info']['keywords']) print("Keywords:", packageInfo["info"]["keywords"])
print('License:',packageInfo['info']['license']) print("License:", packageInfo["info"]["license"])
if packageInfo['info']['requires_dist']: if packageInfo["info"]["requires_dist"]:
print('Dependent Library:',', '.join(packageInfo['info']['requires_dist'])) print("Dependent Library:", ", ".join(packageInfo["info"]["requires_dist"]))
elif allinfo: elif allinfo:
for k,v in packageInfo['info'].items(): for k, v in packageInfo["info"].items():
print(f'{k}: {v}') print(f"{k}: {v}")

99
pmpt/source.py
View File

@ -4,6 +4,7 @@ import os
from rich.table import Table from rich.table import Table
from . import util from . import util
def line_search(li, val, key): def line_search(li, val, key):
# 线性搜索算法 # 线性搜索算法
n = 0 n = 0
@ -14,89 +15,101 @@ def line_search(li, val,key):
n += 1 n += 1
return None return None
def add(url,priority):
'''
添加源
'''
sourceList = jsons.load(open(os.path.join(util.dirs.user_config_dir,'Source.json'))) # 加载source源
if not line_search(sourceList,url,'url'): # 判断源是否存在 def add(url, priority):
util.console.print('❌ [red]The source already exists[/red]') """
添加源
"""
sourceList = jsons.load(
open(os.path.join(util.dirs.user_config_dir, "Source.json"))
) # 加载source源
if not line_search(sourceList, url, "url"): # 判断源是否存在
util.console.print("❌ [red]The source already exists[/red]")
exit(1) exit(1)
sourceList.append({ sourceList.append({"url": url, "id": str(uuid.uuid4())[:8], "priority": priority})
'url':url, jsons.dump(
'id':str(uuid.uuid4())[:8], sourceList, open(os.path.join(util.dirs.user_config_dir, "Source.json"), "w")
'priority':priority )
})
jsons.dump(sourceList,open(os.path.join(util.dirs.user_config_dir,'Source.json'),'w'))
def lists(): def lists():
''' """
列出所有源 列出所有源
''' """
sourceList = jsons.load(open(os.path.join(util.dirs.user_config_dir,'Source.json'))) sourceList = jsons.load(
open(os.path.join(util.dirs.user_config_dir, "Source.json"))
)
# 构建table # 构建table
table = Table(show_header=True) table = Table(show_header=True)
table.add_column("ID", width=6) table.add_column("ID", width=6)
table.add_column('Priority',width=8) table.add_column("Priority", width=8)
table.add_column('URL',width=25) table.add_column("URL", width=25)
# 遍历source列表 # 遍历source列表
for i in sourceList: for i in sourceList:
table.add_row( table.add_row(i["id"], str(i["priority"]), i["url"])
i['id'],
str(i['priority']),
i['url']
)
util.console.print(table) util.console.print(table)
def remove(ids, yes): def remove(ids, yes):
''' """
删除 删除
''' """
if not ids: if not ids:
lists() lists()
ids = util.console.input('Please enter the source sequence number to be deleted:') ids = util.console.input(
"Please enter the source sequence number to be deleted:"
)
sourceList:list = jsons.load(open(os.path.join(util.dirs.user_config_dir,'Source.json'))) sourceList: list = jsons.load(
source,n = line_search(sourceList,ids,'id') open(os.path.join(util.dirs.user_config_dir, "Source.json"))
)
source, n = line_search(sourceList, ids, "id")
if not source: if not source:
util.console.print('❌ [red]The source for this ID does not exist[/red]') util.console.print("❌ [red]The source for this ID does not exist[/red]")
while True: # 是否允许安装 while True: # 是否允许安装
if yes: if yes:
break break
ye = util.console.input('Are you sure you want to delete? [Y/n]: ') ye = util.console.input("Are you sure you want to delete? [Y/n]: ")
if ye.lower() == 'y': if ye.lower() == "y":
break break
elif ye.lower() == 'n': elif ye.lower() == "n":
util.console.print('🛑 [red]The user exited the program[/red]') util.console.print("🛑 [red]The user exited the program[/red]")
exit(1) exit(1)
else: else:
continue continue
sourceList.remove(source) sourceList.remove(source)
jsons.dump(sourceList,open(os.path.join(util.dirs.user_config_dir,'Source.json'),'w')) jsons.dump(
sourceList, open(os.path.join(util.dirs.user_config_dir, "Source.json"), "w")
)
def modify(ids, key, val): def modify(ids, key, val):
''' """
修改 修改
''' """
sourceList:list = jsons.load(open(os.path.join(util.dirs.user_config_dir,'Source.json'))) sourceList: list = jsons.load(
source,n = line_search(sourceList,ids,'id') open(os.path.join(util.dirs.user_config_dir, "Source.json"))
)
source, n = line_search(sourceList, ids, "id")
if not source: if not source:
util.console.print('❌ [red]The source for this ID does not exist[/red]') util.console.print("❌ [red]The source for this ID does not exist[/red]")
if key not in ['url','priority']: if key not in ["url", "priority"]:
util.console.print('❌ [red]The item is not allowed to be modified or does not exist.[/red]') util.console.print(
"❌ [red]The item is not allowed to be modified or does not exist.[/red]"
)
sourceList[n][key] = val sourceList[n][key] = val
jsons.dump(sourceList,open(os.path.join(util.dirs.user_config_dir,'Source.json'),'w')) jsons.dump(
sourceList, open(os.path.join(util.dirs.user_config_dir, "Source.json"), "w")
)

49
pmpt/update.py
View File

@ -5,56 +5,63 @@ import dill
import os import os
from .util import dirs, console from .util import dirs, console
def getSourceID(url): def getSourceID(url):
''' """
获取源id 获取源id
''' """
return url['id'] # 取前八位 return url["id"] # 取前八位
class Index: class Index:
def __init__(self, indexurl): def __init__(self, indexurl):
self.packageList = {} self.packageList = {}
self.IndexURL = indexurl['url'] self.IndexURL = indexurl["url"]
self.number = 0 self.number = 0
self.priority = indexurl['priority'] self.priority = indexurl["priority"]
def addPackage(self, name): def addPackage(self, name):
self.packageList[name] = '1' self.packageList[name] = "1"
self.number += 1 self.number += 1
def getIndex(url): def getIndex(url):
req = requests.get(url['url']) # 请求HTML req = requests.get(url["url"]) # 请求HTML
HTMLIndex = req.text HTMLIndex = req.text
ClassIndex = Index(url) ClassIndex = Index(url)
console.print('🔍 Parsing HTML index...') console.print("🔍 Parsing HTML index...")
for line in tqdm(HTMLIndex.split('\n')): for line in tqdm(HTMLIndex.split("\n")):
# 提取并筛选标签 # 提取并筛选标签
line_list = line.split('>') line_list = line.split(">")
if len(line_list) > 1 and '<a ' in line: if len(line_list) > 1 and "<a " in line:
package_name = line_list[1].split('<')[0] package_name = line_list[1].split("<")[0]
ClassIndex.addPackage(package_name) # 添加包 ClassIndex.addPackage(package_name) # 添加包
console.print('Total number of packages:', str(ClassIndex.number)) console.print("Total number of packages:", str(ClassIndex.number))
console.print('📚 Saving index..."') console.print('📚 Saving index..."')
dill.dump(ClassIndex,open(f'{dirs.user_data_dir}/Index/{getSourceID(url)}.pidx','wb')) dill.dump(
ClassIndex, open(f"{dirs.user_data_dir}/Index/{getSourceID(url)}.pidx", "wb")
)
def getAllIndex(): def getAllIndex():
''' """
SourceList = [ SourceList = [
'https://pypi.tuna.tsinghua.edu.cn/simple', 'https://pypi.tuna.tsinghua.edu.cn/simple',
'https://mirrors.bfsu.edu.cn/pypi/web/simple/' 'https://mirrors.bfsu.edu.cn/pypi/web/simple/'
] ]
''' """
SourceList = jsons.load(open(os.path.join(dirs.user_config_dir,'Source.json'))) # 加载源列表 SourceList = jsons.load(
open(os.path.join(dirs.user_config_dir, "Source.json"))
) # 加载源列表
if len(SourceList) < 1: if len(SourceList) < 1:
console.print('❌ [red]You have not configured any sources.[/red]') console.print("❌ [red]You have not configured any sources.[/red]")
exit(1) exit(1)
for source in SourceList: # 遍历源列表 for source in SourceList: # 遍历源列表
console.print('📚 Downloading index from', source['url']+'...') console.print("📚 Downloading index from", source["url"] + "...")
getIndex(source) getIndex(source)
console.print('✅ [green]Index downloaded successfully![/green]') console.print("✅ [green]Index downloaded successfully![/green]")

93
pmpt/util.py
View File

@ -15,98 +15,119 @@ dirs = PlatformDirs("PMPT", "MoYan")
IndexList = [] IndexList = []
console = Console() console = Console()
def getVer(baseVar): def getVer(baseVar):
baseVar = baseVar # + '.' + os.environ.get('GITHUB_RUN_ID', str(int(time.time()))[:6]) baseVar = (
logger.info('PMPT '+baseVar) baseVar # + '.' + os.environ.get('GITHUB_RUN_ID', str(int(time.time()))[:6])
)
logger.info("PMPT " + baseVar)
return baseVar return baseVar
def GlobalDecorator(frame, event, arg): def GlobalDecorator(frame, event, arg):
if event == 'call': if event == "call":
try: try:
func_name = frame.f_code.co_name func_name = frame.f_code.co_name
module_name = frame.f_globals['__name__'] module_name = frame.f_globals["__name__"]
package_name = module_name.split('.')[0] # 假设包名为模块名的第一部分 package_name = module_name.split(".")[0] # 假设包名为模块名的第一部分
if package_name == 'pmpt': if package_name == "pmpt":
logger.trace(f"调用函数 {module_name}.{func_name}") logger.trace(f"调用函数 {module_name}.{func_name}")
except: except:
pass pass
return GlobalDecorator return GlobalDecorator
sys.settrace(GlobalDecorator) sys.settrace(GlobalDecorator)
logger.remove() logger.remove()
logger.add( logger.add(
os.path.join(dirs.user_data_dir,'log.log'), os.path.join(dirs.user_data_dir, "log.log"),
level='TRACE', level="TRACE",
) )
__version__ = getVer('1.0.4') __version__ = getVer("1.0.3")
def init(): def init():
os.makedirs(dirs.user_data_dir, exist_ok=True) os.makedirs(dirs.user_data_dir, exist_ok=True)
os.makedirs(os.path.join(dirs.user_data_dir,'Index'),exist_ok=True) os.makedirs(os.path.join(dirs.user_data_dir, "Index"), exist_ok=True)
os.makedirs(dirs.user_config_dir, exist_ok=True) os.makedirs(dirs.user_config_dir, exist_ok=True)
os.makedirs(dirs.user_cache_dir, exist_ok=True) os.makedirs(dirs.user_cache_dir, exist_ok=True)
if not os.path.exists(os.path.join(dirs.user_config_dir,'Source.json')): if not os.path.exists(os.path.join(dirs.user_config_dir, "Source.json")):
open(os.path.join(dirs.user_config_dir,'Source.json'),'w').write('[]') open(os.path.join(dirs.user_config_dir, "Source.json"), "w").write("[]")
from . import source from . import source
source.add() source.add()
if not os.path.exists(os.path.join(dirs.user_config_dir,'api.url')): if not os.path.exists(os.path.join(dirs.user_config_dir, "api.url")):
open(os.path.join(dirs.user_config_dir,'api.url'),'w').write('https://pypi.org/pypi/{}/json') open(os.path.join(dirs.user_config_dir, "api.url"), "w").write(
"https://pypi.org/pypi/{}/json"
)
def bubbleSort(arr): def bubbleSort(arr):
for i in range(1, len(arr)): for i in range(1, len(arr)):
for j in range(0, len(arr) - i): for j in range(0, len(arr) - i):
if arr[j]['priority'] < arr[j+1]['priority']: if arr[j]["priority"] < arr[j + 1]["priority"]:
arr[j], arr[j + 1] = arr[j + 1], arr[j] arr[j], arr[j + 1] = arr[j + 1], arr[j]
return arr return arr
def loadIndex(): def loadIndex():
''' """
加载索引 加载索引
''' """
today = datetime.datetime.today() today = datetime.datetime.today()
if today.month == 4 and today.day == 1: if today.month == 4 and today.day == 1:
console.print('😱😱😱 [red]Oh no, your indexes have all been eaten by a python snake.[/red]') console.print(
"😱😱😱 [red]Oh no, your indexes have all been eaten by a python snake.[/red]"
)
time.sleep(2.3) time.sleep(2.3)
console.print('😄 [green]But after our life-and-death struggle, I managed to retrieve all of your indexes.[/green]') console.print(
sourceList = jsons.load(open(os.path.join(dirs.user_config_dir,'Source.json'))) "😄 [green]But after our life-and-death struggle, I managed to retrieve all of your indexes.[/green]"
)
sourceList = jsons.load(open(os.path.join(dirs.user_config_dir, "Source.json")))
sourceList = bubbleSort(sourceList) sourceList = bubbleSort(sourceList)
for source in sourceList: for source in sourceList:
if not os.path.exists(os.path.join(dirs.user_data_dir,'Index',source['id']+'.pidx')): if not os.path.exists(
logger.warning(source['url']+'没有索引') os.path.join(dirs.user_data_dir, "Index", source["id"] + ".pidx")
console.print(f'⚠️ [yellow]{source["url"]} did not create an index.[/yellow]') ):
logger.warning(source["url"] + "没有索引")
console.print(
f'⚠️ [yellow]{source["url"]} did not create an index.[/yellow]'
)
continue continue
logger.debug(source) logger.debug(source)
IndexFile = dill.load(open(os.path.join(dirs.user_data_dir,'Index',source['id']+'.pidx'),'rb')) # 加载索引 IndexFile = dill.load(
IndexList.append(IndexFile) open(
if len(IndexList) == 0: os.path.join(dirs.user_data_dir, "Index", source["id"] + ".pidx"), "rb"
)
) # 加载索引
yield IndexFile
raise FileNotFoundError('No index. Run "pmpt update" first to update the index')
def runpip(command, other=None, dbg=False, out=True) -> Popen: def runpip(command, other=None, dbg=False, out=True) -> Popen:
''' """
运行pip 运行pip
''' """
logger.trace('调用runpip') logger.trace("调用runpip")
if not other: if not other:
other = [] other = []
baseCommand = [sys.executable,'-m','pip'] baseCommand = [sys.executable, "-m", "pip"]
baseCommand.append(command) baseCommand.append(command)
Command = baseCommand + other Command = baseCommand + other
if dbg: if dbg:
console.print('Command to be run:',' '.join(Command)) console.print("Command to be run:", " ".join(Command))
logger.debug(' ',) logger.debug(
" ",
)
runClass = Popen(Command, stdout=subprocess.PIPE, stderr=subprocess.PIPE) runClass = Popen(Command, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
if out: if out:
for line in iter(runClass.stdout.readline, b''): for line in iter(runClass.stdout.readline, b""):
# 在这里你可以对每一行输出进行处理 # 在这里你可以对每一行输出进行处理
line = line.decode('utf-8').strip() # 将字节转换为字符串并去除换行符 line = line.decode("utf-8").strip() # 将字节转换为字符串并去除换行符
console.print(line) console.print(line)
if runClass.returncode != 0: if runClass.returncode != 0:
console.print(runClass.stderr.read().decode()) console.print(runClass.stderr.read().decode())
@ -114,5 +135,3 @@ def runpip(command,other=None,dbg=False,out=True) -> Popen:
else: else:
runClass.wait() runClass.wait()
return runClass return runClass

0
requirements.txt → requirements/base.txt
View File

0
requirements/web.txt Normal file
View File

36
setup.py
View File

@ -2,31 +2,29 @@ from setuptools import setup, find_packages
from pmpt import util from pmpt import util
setup( setup(
name='pmpt', # 包的名称 name="pmpt", # 包的名称
version=util.__version__, # 版本号 version=util.__version__, # 版本号
packages=find_packages(), # 包含的包 packages=find_packages(), # 包含的包
author='MoYan', # 作者 author="MoYan", # 作者
author_email='moyan@moyanjdc.top', # 作者邮箱 author_email="moyan@moyanjdc.top", # 作者邮箱
description='A Python Package Advanced Manager', # 包的简要描述 description="A Python Package Advanced Manager", # 包的简要描述
long_description=open("readme.md").read(), # 包的详细描述 long_description=open("readme.md").read(), # 包的详细描述
long_description_content_type='text/markdown', # 描述的内容类型 long_description_content_type="text/markdown", # 描述的内容类型
classifiers=[ # 包的分类信息 classifiers=[ # 包的分类信息
'Development Status :: 3 - Alpha', "Development Status :: 3 - Alpha",
'Intended Audience :: Developers', "Intended Audience :: Developers",
'License :: OSI Approved :: MIT License', "License :: OSI Approved :: MIT License",
'Programming Language :: Python :: 3', "Programming Language :: Python :: 3",
'Programming Language :: Python :: 3.7', "Programming Language :: Python :: 3.7",
'Programming Language :: Python :: 3.8', "Programming Language :: Python :: 3.8",
'Programming Language :: Python :: 3.9', "Programming Language :: Python :: 3.9",
], ],
setup_requires = ['platformdirs'], setup_requires=["platformdirs"],
install_requires=open('requirements.txt').read().split('\n'), install_requires=open("requirements/base.txt").read().split("\n"),
entry_points={ entry_points={
'console_scripts': [ "console_scripts": ["pmpt=pmpt:cli"],
'pmpt=pmpt:cli'
],
}, },
extras_require={ extras_require={
'web': ["flask"], "web": open("requirements/web.txt").read().split("\n"),
} },
) )