file-type

JSON API有效载荷资源管理器工具使用与部署指南

下载需积分: 10 | 339KB | 更新于2025-09-10 | 57 浏览量 | 0 下载量 举报 收藏
download 立即下载
JSON API是一种网络数据交互的格式,它定义了一种高效且可扩展的机制,使得客户端与服务器之间的数据传输更加简洁和结构化。JSON API格式允许客户端明确地指定其需要数据的形状,并且允许服务器仅发送那些数据,从而提高了应用程序的性能和可维护性。在IT行业,JSON API被广泛用于RESTful Web服务中。 jsonapi-payload-explorer是一个JavaScript编写的工具,旨在帮助开发者预览、搜索和检查JSON API格式的有效载荷。有效载荷指的是在网络传输过程中数据的部分,它包含了实际的信息内容。这个工具特别适合在开发过程中调试和验证数据的正确性。开发者可以利用这个工具来确保他们的应用程序正确地处理JSON API数据。 使用jsonapi-payload-explorer之前,开发者需要先通过git进行项目的克隆。Git是一个开源的分布式版本控制系统,它允许开发者在不同的开发环境中协同工作,并且能够有效地跟踪项目代码的变更历史。在这个案例中,开发者需要通过SSH协议使用git clone命令来从GitHub上的远程仓库下载jsonapi-payload-explorer项目到本地。 完成克隆后,开发者需要切换到项目目录中(使用cd命令),然后通过yarn install命令安装项目依赖。Yarn是一个快速、可靠和安全的依赖管理工具,它允许开发者定义项目需要的依赖,并且会自动下载这些依赖到本地。一旦依赖安装完毕,开发者就可以通过执行yarn start命令来启动项目。 项目的启动过程中,开发者可以查看工具如何展示JSON API格式的有效载荷,同时也能够进行搜索和检查。这对于开发者来说是一个宝贵的调试资源,因为JSON API的有效载荷需要精确地遵循JSON API规范。比如,在JSON API中,数据通常以数组的形式出现在响应的data键下,同时还需要定义相关联的资源类型和资源标识符。 部署jsonapi-payload-explorer的过程非常简单,开发者可以通过执行yarn run deploy命令将应用程序部署到GitHub页面。这个过程实际上是利用GitHub提供的静态网页托管服务,将应用的构建产物推送上去。GitHub Pages是一个免费的静态网站托管服务,允许开发者直接从GitHub仓库部署静态网站。这使得开发者可以快速地将他们的工具和项目页面分享给他人。 在标签方面,"json"指的是JavaScript对象表示法,它是一种轻量级的数据交换格式。"jsonapi"指的是JSON API规范,即我们正在讨论的数据格式。"payloads"则指的是在数据传输过程中被携带的数据部分。而"JavaScript"是jsonapi-payload-explorer工具的编程语言。 压缩包子文件的文件名称列表中的"jsonapi-payload-explorer-master"表示的是这个项目的主要分支,通常master分支代表的是项目的稳定版本。这个命名方式符合开源项目的常规命名习惯,便于其他开发者理解和使用。 总结而言,jsonapi-payload-explorer是一款能够帮助开发者管理和检查JSON API数据的工具,它通过简洁的命令行操作,允许开发者快速预览、搜索和验证JSON格式的有效载荷。此外,通过git、yarn以及GitHub Pages的使用,开发者能够轻松地维护和部署这个工具,使得整个开发流程更加高效。

相关推荐

filetype

import json import os import logging from datetime import datetime, timedelta import pytz import requests from urllib.parse import quote import time from selenium import webdriver from selenium.webdriver.edge.service import Service from selenium.webdriver.common.by import By from selenium.webdriver.edge.options import Options from selenium.webdriver.support.ui import WebDriverWait from selenium.webdriver.support import expected_conditions as EC from bs4 import BeautifulSoup import re import shutil import glob import asyncio import edge_tts # 配置日志 logging.basicConfig( level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s', handlers=[ logging.FileHandler("data_fetcher.log"), logging.StreamHandler() ] ) logger = logging.getLogger(__name__) # 下载目录配置 DOWNLOAD_DIR = os.path.expanduser("~/Downloads") # 浏览器默认下载目录 TARGET_DIR = "./bobao" # 目标保存目录(当前目录下的baobao文件夹) def get_broadcast_data_with_token(tenant_access_token): """使用token获取飞书的数据""" url = 'https://open.feishu.cn/open-apis/bitable/v1/apps/E1zybPqiqa0TaesZjKKch5ZcnJd/tables/tblwFY4k3pmrV5WK/records/search' headers = { 'Content-Type': 'application/json', 'Authorization': f'Bearer {tenant_access_token}' } data = {} try: logger.info(f"正在请求飞书数据,URL: {url}") response = requests.post(url, headers=headers, json=data, timeout=30) response.raise_for_status() response_dict = response.json() items = response_dict.get("data", {}).get("items", []) logger.info(f"成功获取飞书数据,共 {len(items)} 条记录") data = [] for item in items: fields = item.get("fields", {}) time_segment = extract_time_segment(fields, '时间段') song1_text = extract_text_field(fields, '壹歌曲-歌手') song2_text = extract_text_field(fields, '贰歌曲-歌手') # 新增:从文本字段获取播报内容 song1_broadcast = extract_text_field(fields, '需更新文案-播报') song2_broadcast = extract_text_field(fields, '需更新文案2-播报') if time_segment == "08:10-08:15": song1_name = "" song1_artist = "" song2_name = "" song2_artist = "" else: song1_parts = song1_text.split(' ', 1) song1_name = song1_parts[0] if len(song1_parts) > 0 else "" song1_artist = song1_parts[1] if len(song1_parts) > 1 else "" song2_parts = song2_text.split(' ', 1) song2_name = song2_parts[0] if len(song2_parts) > 0 else "" song2_artist = song2_parts[1] if len(song2_parts) > 1 else "" data.append({ "播音日期": extract_broadcast_date(fields, '播音日期'), "时间段": time_segment, "开播音乐file_token": extract_file_token(fields, '开播音乐'), "开场白-播报file_token": extract_file_token(fields, '开场白-播报'), "壹歌曲-歌手": song1_text, "需更新文案-播报": song1_broadcast, # 更新为文本内容 "贰歌曲-歌手": song2_text, "需更新文案2-播报": song2_broadcast, # 更新为文本内容 "壹歌名": song1_name, "壹歌手": song1_artist, "贰歌名": song2_name, "贰歌手": song2_artist, "结束语-播报file_token": extract_file_token(fields, '结束语-播报'), "结束音乐file_token": extract_file_token(fields, '结束音乐') }) return data except requests.exceptions.HTTPError as http_err: logger.error(f"HTTP 错误发生: {http_err}") except requests.exceptions.Timeout: logger.error("请求超时,服务器响应时间过长") except requests.exceptions.ConnectionError: logger.error("连接错误,无法连接到服务器") except Exception as err: logger.error(f"其他错误发生: {err}", exc_info=True) return [] def extract_file_token(fields, field_name): """提取 file_token""" field_data = fields.get(field_name, []) if isinstance(field_data, list) and len(field_data) > 0: value = field_data[0] if isinstance(value, dict): return value.get("file_token", "") return '' def extract_text_field(fields, field_name): """提取文本字段内容""" field_data = fields.get(field_name, []) if isinstance(field_data, list) and len(field_data) > 0: value = field_data[0] if isinstance(value, dict): # 尝试获取text字段,如果没有则获取content return value.get("text", value.get("content", "")) return '' def extract_time_segment(fields, field_name): """提取时间段字段""" field_data = fields.get(field_name, []) if isinstance(field_data, list) and len(field_data) > 0: value = field_data[0] if isinstance(value, dict): return value.get("text", "") return None def extract_broadcast_date(fields, field_name): """提取播音日期字段""" field_data = fields.get(field_name, 0) if isinstance(field_data, int): try: timestamp = field_data / 1000 parsed_date = datetime.fromtimestamp(timestamp, tz=pytz.utc).astimezone(pytz.timezone('Asia/Shanghai')) return parsed_date.strftime("%Y-%m-%d") except (ValueError, OverflowError): pass return None def get_auth_token(): """获取认证 token""" url = "https://open.feishu.cn/open-apis/auth/v3/tenant_access_token/internal" headers = {"Content-Type": "application/json; charset=utf-8"} payload = {"app_id": "cli_a882683e8779d00c", "app_secret": "3NKkALA7vyMRVnpKJinmrb1LJ7YuK4H0"} try: logger.info("正在获取认证token") response = requests.post(url, json=payload, headers=headers, timeout=30) response.raise_for_status() data = response.json() if data["code"] == 0: logger.info("成功获取认证token") return data["tenant_access_token"] else: logger.error(f"请求失败:{data['msg']}(错误码:{data['code']})") except requests.exceptions.HTTPError as http_err: logger.error(f"HTTP 错误发生: {http_err}") except requests.exceptions.Timeout: logger.error("获取token超时") except requests.exceptions.ConnectionError: logger.error("连接错误,无法获取token") except Exception as e: logger.error(f"获取token异常:{e}", exc_info=True) return None def display_data(data): """展示获取的数据""" if not data: print("没有获取到任何数据") return print(f"共获取到 {len(data)} 条广播数据记录:") for i, item in enumerate(data, 1): print(f"\n--- 记录 {i} ---") for key, value in item.items(): if key.endswith('file_token') and value: print(f"{key}: {value[:10]}...") else: print(f"{key}: {value}") def format_song_info(record): """格式化并打印歌曲信息""" print("\n歌曲信息:") print(f"壹歌名: {record['壹歌名']}") print(f"壹歌手: {record['壹歌手']}") print(f"贰歌名: {record['贰歌名']}") print(f"贰歌手: {record['贰歌手']}") # 新增:打印播报内容 print("\n播报内容:") print(f"需更新文案-播报: {record['需更新文案-播报']}") print(f"需更新文案2-播报: {record['需更新文案2-播报']}") def wait_for_download(initial_files, timeout=60): """等待下载完成并返回新文件路径""" start_time = time.time() while time.time() - start_time < timeout: current_files = os.listdir(DOWNLOAD_DIR) new_files = [f for f in current_files if f not in initial_files] # 过滤掉临时下载文件 valid_files = [f for f in new_files if not f.endswith('.crdownload')] if valid_files: # 返回第一个找到的有效文件 return os.path.join(DOWNLOAD_DIR, valid_files[0]) time.sleep(1) print(f"等待下载超时({timeout}秒)") return None def download_song(song_name, artist=None, save_path=TARGET_DIR, custom_filename=None): """ 搜索并下载指定歌曲,下载后重命名为指定名称 参数: song_name (str): 歌曲名称 artist (str, optional): 歌手名称,默认为None save_path (str, optional): 保存路径,默认为"./bobao" custom_filename (str, optional): 自定义文件名,默认为None """ # 创建保存目录(如果不存在) if not os.path.exists(save_path): os.makedirs(save_path) # 构建搜索关键词 search_query = song_name if artist: search_query += f" {artist}" print(f"正在搜索歌曲: {search_query}") # 记录下载前的文件列表 initial_files = os.listdir(DOWNLOAD_DIR) # 设置Edge浏览器选项 edge_options = Options() edge_options.add_argument('--disable-gpu') edge_options.add_argument('--no-sandbox') edge_options.add_argument('--disable-dev-shm-usage') # 指定EdgeDriver的路径 edge_driver_path = r"C:\Users\shaopeng.qi\Downloads\edgedriver_win64 (1)\msedgedriver.exe" # 检查路径是否存在 if not os.path.exists(edge_driver_path): print(f"错误: EdgeDriver路径不存在 - {edge_driver_path}") print("请下载与您Edge浏览器版本匹配的EdgeDriver,并更新路径") print("下载地址: https://developer.microsoft.com/en-us/microsoft-edge/tools/webdriver/") return # 初始化WebDriver service = Service(executable_path=edge_driver_path) service.log_path = "edgedriver.log" driver = webdriver.Edge(service=service, options=edge_options) try: # 直接访问搜索结果页面 search_url = f"https://www.gequbao.com/s/{quote(search_query)}.html" print(f"正在访问搜索结果页面: {search_url}") driver.get(search_url) # 等待页面加载完成 WebDriverWait(driver, 10).until( EC.presence_of_element_located((By.CSS_SELECTOR, ".card-body")) ) # 检查是否有搜索结果 try: no_results = WebDriverWait(driver, 5).until( EC.presence_of_element_located((By.XPATH, "//div[contains(text(), '没有找到相关歌曲')]")) ) print("未找到匹配的歌曲") return except: # 没有找到"没有找到相关歌曲"的提示,继续执行 pass # 滚动到页面底部,确保所有元素都加载 driver.execute_script("window.scrollTo(0, document.body.scrollHeight);") time.sleep(1) # 查找第一个歌曲的链接 print("正在查找歌曲链接...") song_links = WebDriverWait(driver, 10).until( EC.presence_of_all_elements_located((By.CSS_SELECTOR, ".music-link")) ) if not song_links: print("未找到歌曲链接") with open("error_page.html", "w", encoding="utf-8") as f: f.write(driver.page_source) print("已保存当前页面源码到error_page.html,可用于分析问题") return # 获取第一个歌曲链接的href属性 first_song_link = song_links[0].get_attribute("href") print(f"找到歌曲链接: {first_song_link}") # 提取歌曲ID match = re.search(r'/music/(\d+)', first_song_link) if not match: print("无法从链接中提取歌曲ID") with open("error_page.html", "w", encoding="utf-8") as f: f.write(driver.page_source) print("已保存当前页面源码到error_page.html,可用于分析问题") return song_id = match.group(1) print(f"提取的歌曲ID: {song_id}") # 直接构建并访问歌曲详情页URL song_detail_url = f"https://www.gequbao.com/music/{song_id}" print(f"正在访问歌曲详情页: {song_detail_url}") # 打开新窗口 driver.execute_script(f"window.open('{song_detail_url}');") # 等待新窗口打开 time.sleep(2) # 检查是否打开了新窗口 if len(driver.window_handles) > 1: # 切换到新窗口 for window in driver.window_handles: if window != driver.current_window_handle: driver.switch_to.window(window) print("已切换到歌曲详情页") break else: print("未成功打开歌曲详情页,尝试直接访问") # 直接访问歌曲详情页 driver.get(song_detail_url) print("已直接访问歌曲详情页") # 等待页面加载 print("等待页面加载完成...") time.sleep(5) # 检查是否在歌曲详情页 try: # 查找歌曲标题元素,确认在详情页 song_title = WebDriverWait(driver, 5).until( EC.presence_of_element_located((By.CSS_SELECTOR, ".song-title")) ) print(f"已进入歌曲详情页,歌曲标题: {song_title.text}") except: # 如果找不到歌曲标题元素,尝试其他可能的元素 try: # 尝试查找下载按钮 download_btn = WebDriverWait(driver, 3).until( EC.presence_of_element_located((By.XPATH, "//a[contains(text(), '下载')]")) ) print("已进入歌曲详情页,找到下载按钮") except: print("无法确认是否在歌曲详情页") with open("error_page.html", "w", encoding="utf-8") as f: f.write(driver.page_source) print("已保存当前页面源码到error_page.html,可用于分析问题") return # 点击下载按钮 print("正在查找并点击下载按钮...") try: download_btn = WebDriverWait(driver, 10).until( EC.element_to_be_clickable((By.XPATH, "//*[@id=\"btn-download-mp3\"]")) ) print("找到下载按钮") # 点击下载按钮 download_btn.click() print("已点击下载按钮") # 等待下载对话框出现 print("等待下载对话框出现...") time.sleep(3) # 点击低品质MP3按钮 - 使用更灵活的XPath选择器 print("正在查找并点击低品质MP3按钮...") try: # 使用通配符匹配动态ID部分 low_quality_btn = WebDriverWait(driver, 10).until( EC.element_to_be_clickable((By.XPATH, "//*[starts-with(@id, 'jconfirm-box')]/div/ul/li[2]/a")) ) print("找到低品质MP3按钮") # 点击低品质MP3按钮 low_quality_btn.click() # 构建保存文件名 if custom_filename: target_filename = f"{custom_filename}.mp3" elif artist: target_filename = f"{artist} - {song_name}.mp3" else: target_filename = f"{song_name}.mp3" target_path = os.path.join(save_path, target_filename) print(f"已点击下载按钮,等待下载完成后将保存为: {target_path}") # 等待下载完成 print("等待下载开始...") downloaded_file = wait_for_download(initial_files, timeout=60) if downloaded_file: # 等待文件完全下载(避免文件正在写入时移动) time.sleep(5) # 移动并重命名文件 try: shutil.move(downloaded_file, target_path) print(f"歌曲已成功下载并重命名为: {target_path}") except Exception as move_err: print(f"移动文件时出错: {move_err}") print(f"下载的文件保存在: {downloaded_file}") else: print("未检测到新的下载文件") except Exception as e: print(f"点击低品质MP3按钮时出错: {e}") with open("error_page.html", "w", encoding="utf-8") as f: f.write(driver.page_source) print("已保存当前页面源码到error_page.html,可用于分析问题") return except Exception as e: print(f"点击下载按钮时出错: {e}") with open("error_page.html", "w", encoding="utf-8") as f: f.write(driver.page_source) print("已保存当前页面源码到error_page.html,可用于分析问题") return except Exception as e: print(f"操作过程中发生错误: {e}") with open("error_page.html", "w", encoding="utf-8") as f: f.write(driver.page_source) print("已保存当前页面源码到error_page.html,可用于分析问题") finally: # 关闭浏览器 driver.quit() print("浏览器已关闭") def download_required_songs(data): """下载指定的两首歌曲并命名为file_4和file_6""" if not data or len(data) == 0: print("没有可下载的歌曲数据") return first_record = data[0] # 下载第一首歌 (file_4) song1_name = first_record.get("壹歌名") song1_artist = first_record.get("壹歌手") if song1_name and song1_artist: print(f"\n准备下载第一首歌: {song1_name} - {song1_artist},保存为 file_4.mp3") try: download_song(song1_name, song1_artist, custom_filename="file_4") except Exception as e: print(f"下载失败: {song1_name} - {song1_artist}, 错误: {e}") else: print("第一首歌信息不完整,无法下载") print(f"歌名: {song1_name}, 歌手: {song1_artist}") # 下载第二首歌 (file_6) song2_name = first_record.get("贰歌名") song2_artist = first_record.get("贰歌手") if song2_name and song2_artist: print(f"\n准备下载第二首歌: {song2_name} - {song2_artist},保存为 file_6.mp3") try: download_song(song2_name, song2_artist, custom_filename="file_6") except Exception as e: print(f"下载失败: {song2_name} - {song2_artist}, 错误: {e}") else: print("第二首歌信息不完整,无法下载") print(f"歌名: {song2_name}, 歌手: {song2_artist}") async def convert_text_to_speech(text, output_file): """将文本转换为语音并保存为MP3文件""" if text: try: communicate = edge_tts.Communicate(text, voice="zh-CN-YunyangNeural") await communicate.save(output_file) print(f"语音已保存为:{output_file}") return True except Exception as e: print(f"文本转语音失败: {e}") return False else: print(f"文本为空,跳过转换: {output_file}") return False async def convert_broadcast_texts_async(data): """异步转换播报文本为语音文件""" if not data or len(data) == 0: print("没有可转换的播报文本数据") return first_record = data[0] # 确保目标目录存在 if not os.path.exists(TARGET_DIR): os.makedirs(TARGET_DIR) # 转换第一个播报文本 (file_3) text1 = first_record.get("需更新文案-播报", "") output_file1 = os.path.join(TARGET_DIR, "file_3.mp3") print(f"\n准备转换第一个播报文本,保存为 file_3.mp3") print(f"文本内容: {text1[:50]}...") # 只显示前50个字符 success1 = await convert_text_to_speech(text1, output_file1) if success1: print("第一个播报文本转换成功") # 转换第二个播报文本 (file_5) text2 = first_record.get("需更新文案2-播报", "") output_file2 = os.path.join(TARGET_DIR, "file_5.mp3") print(f"\n准备转换第二个播报文本,保存为 file_5.mp3") print(f"文本内容: {text2[:50]}...") # 只显示前50个字符 success2 = await convert_text_to_speech(text2, output_file2) if success2: print("第二个播报文本转换成功") def convert_broadcast_texts(data): """转换播报文本为语音文件 - 兼容同步和异步环境""" try: # 尝试获取当前事件循环 loop = asyncio.get_event_loop() except RuntimeError: # 如果没有活动的事件循环,创建一个新的 loop = asyncio.new_event_loop() asyncio.set_event_loop(loop) # 检查事件循环是否正在运行 if loop.is_running(): # 如果事件循环已经在运行(如在Jupyter中),使用asyncio.run_coroutine_threadsafe # 注意:这需要在支持多线程的环境中 future = asyncio.run_coroutine_threadsafe(convert_broadcast_texts_async(data), loop) future.result() # 等待异步任务完成 else: # 如果事件循环没有运行,直接运行 loop.run_until_complete(convert_broadcast_texts_async(data)) def main(): """主函数""" logger.info("===== 飞书数据获取及媒体处理程序启动 =====") # 获取认证token authorization = get_auth_token() if not authorization: logger.error("获取认证token失败,程序退出") return # 获取广播数据 data = get_broadcast_data_with_token(authorization) # 展示获取的数据 display_data(data) # 保存数据到JSON文件 try: with open('broadcast_data.json', 'w', encoding='utf-8') as f: json.dump(data, f, ensure_ascii=False, indent=2) print("\n数据已保存到 broadcast_data.json 文件") except Exception as e: logger.error(f"保存数据失败: {e}") # 打印歌曲信息 if data: format_song_info(data[0]) # 自动下载两首歌曲,命名为file_4和file_6 download_required_songs(data) # 转换播报文本为语音文件 convert_broadcast_texts(data) logger.info("===== 飞书数据获取及媒体处理程序结束 =====") if __name__ == "__main__": main() 检查上述代码中存在的问题

林文曦
  • 粉丝: 46
上传资源 快速赚钱