同步腾讯云证书 遍历目录下的nginx证书压缩包，解压并同步到面板ssl目录
(tencent_ssl_path="/root/tencent_ssl")
| 697 | exit(public.version()) |
| 698 | |
| 699 | def sync_tencent_ssl(tencent_ssl_path="/root/tencent_ssl"): |
| 700 | """同步腾讯云证书 |
| 701 | 遍历目录下的nginx证书压缩包，解压并同步到面板ssl目录 |
| 702 | """ |
| 703 | import os |
| 704 | import shutil |
| 705 | import panelSSL |
| 706 | from sslModel import certModel |
| 707 | ss = panelSSL.panelSSL() |
| 708 | |
| 709 | cert_main = certModel.main() |
| 710 | |
| 711 | panel_ssl_path = "/www/server/panel/vhost/ssl" |
| 712 | |
| 713 | try: |
| 714 | # 确保源目录存在 |
| 715 | if not os.path.exists(tencent_ssl_path): |
| 716 | print(f"目录不存在: {tencent_ssl_path}") |
| 717 | return False |
| 718 | # 获取所有网站和对应的域名 |
| 719 | all_sites = public.M('sites').field('name,id').select() |
| 720 | if not all_sites: |
| 721 | print("没有找到任何网站信息") |
| 722 | return False |
| 723 | for site in all_sites: |
| 724 | domians = public.M('domain').where("pid=?", (site["id"],)).field('name').select() |
| 725 | if not domians: |
| 726 | site["domains"] = [] |
| 727 | site["domains"] = [domain["name"] for domain in domians] |
| 728 | BatchInfo = [] |
| 729 | |
| 730 | # 遍历目录下的所有文件 |
| 731 | for filename in os.listdir(tencent_ssl_path): |
| 732 | if not filename.endswith('_nginx.zip'): |
| 733 | continue |
| 734 | |
| 735 | # 获取域名（去除_nginx.zip后缀） |
| 736 | domain = filename.replace('_nginx.zip', '') |
| 737 | |
| 738 | zip_path = os.path.join(tencent_ssl_path, filename) |
| 739 | extract_path = os.path.join(tencent_ssl_path, domain + '_nginx') |
| 740 | ssl_domain_path = os.path.join(panel_ssl_path, domain) |
| 741 | |
| 742 | try: |
| 743 | # 解压文件 |
| 744 | public.ExecShell("cd {} && unzip {}".format(tencent_ssl_path, zip_path)) |
| 745 | |
| 746 | # 创建目标ssl目录 |
| 747 | os.makedirs(ssl_domain_path, exist_ok=True) |
| 748 | |
| 749 | # 复制证书文件 |
| 750 | bundle_src = os.path.join(extract_path, f"{domain}_bundle.pem") |
| 751 | key_src = os.path.join(extract_path, f"{domain}.key") |
| 752 | get = public.to_dict_obj({}) |
| 753 | get.key = public.readFile(key_src) |
| 754 | get.csr = public.readFile(bundle_src) |
| 755 | res=cert_main.save_cert(get) |
| 756 | if res.get("status") == True: |
no test coverage detected