3a63bc907e
- 新增 /api/fetch-skill-info 接口识别 GitHub skills 仓库 - 解析 skill markdown 的简介、应用场景和包含内容 - Skills 创建页支持一键自动填充标题、简介、来源链接 Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
400 lines
14 KiB
Python
400 lines
14 KiB
Python
# -*- coding: utf-8 -*-
|
||
"""
|
||
网站信息抓取工具
|
||
"""
|
||
import requests
|
||
from bs4 import BeautifulSoup
|
||
from urllib.parse import urljoin, urlparse
|
||
import os
|
||
import re
|
||
from PIL import Image
|
||
from io import BytesIO
|
||
|
||
class WebsiteFetcher:
|
||
"""网站信息抓取器"""
|
||
|
||
def __init__(self, timeout=10):
|
||
self.timeout = timeout
|
||
self.headers = {
|
||
'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36'
|
||
}
|
||
|
||
def fetch_website_info(self, url):
|
||
"""
|
||
抓取网站信息
|
||
|
||
Args:
|
||
url: 网站URL
|
||
|
||
Returns:
|
||
dict: 包含name, description, logo_url的字典，失败返回None
|
||
"""
|
||
try:
|
||
# 确保URL包含协议
|
||
if not url.startswith(('http://', 'https://')):
|
||
url = 'https://' + url
|
||
|
||
# 请求网页
|
||
response = requests.get(url, headers=self.headers, timeout=self.timeout, allow_redirects=True)
|
||
response.raise_for_status()
|
||
response.encoding = response.apparent_encoding # 自动检测编码
|
||
|
||
# 解析HTML
|
||
soup = BeautifulSoup(response.text, 'html.parser')
|
||
|
||
# 提取信息
|
||
info = {
|
||
'name': self._extract_title(soup),
|
||
'description': self._extract_description(soup),
|
||
'logo_url': self._extract_logo(soup, url)
|
||
}
|
||
|
||
return info
|
||
|
||
except Exception as e:
|
||
print(f"抓取网站信息失败: {str(e)}")
|
||
return None
|
||
|
||
def fetch_skill_info(self, url):
|
||
"""
|
||
抓取 Skill 信息（GitHub 仓库）
|
||
|
||
Args:
|
||
url: GitHub 仓库URL
|
||
|
||
Returns:
|
||
dict: 包含name, short_desc, github_url, source_repo的字典
|
||
"""
|
||
try:
|
||
# 解析 GitHub URL
|
||
# 支持格式: https://github.com/user/repo 或 https://github.com/user/repo/tree/main
|
||
parsed = urlparse(url)
|
||
if 'github.com' not in parsed.netloc:
|
||
return None
|
||
|
||
path_parts = [p for p in parsed.path.split('/') if p]
|
||
if len(path_parts) < 2:
|
||
return None
|
||
|
||
owner = path_parts[0]
|
||
repo = path_parts[1].replace('.git', '')
|
||
|
||
# 获取仓库信息
|
||
api_url = f"https://api.github.com/repos/{owner}/{repo}"
|
||
response = requests.get(api_url, headers=self.headers, timeout=self.timeout)
|
||
if response.status_code != 200:
|
||
return None
|
||
|
||
repo_info = response.json()
|
||
|
||
# 获取 .claude 目录下的 skills
|
||
skills_url = f"https://api.github.com/repos/{owner}/{repo}/contents/.claude/skills"
|
||
skills_response = requests.get(skills_url, headers=self.headers, timeout=self.timeout)
|
||
|
||
skills = []
|
||
if skills_response.status_code == 200:
|
||
skills_data = skills_response.json()
|
||
for item in skills_data:
|
||
if item.get('type') == 'file' and item['name'].endswith('.md'):
|
||
# 获取 skill 文件内容
|
||
skill_content = requests.get(item['download_url'], headers=self.headers, timeout=self.timeout)
|
||
if skill_content.status_code == 200:
|
||
skill_info = self._parse_skill_file(
|
||
item['name'],
|
||
skill_content.text,
|
||
item.get('html_url') or item.get('download_url') or ''
|
||
)
|
||
if skill_info:
|
||
skills.append(skill_info)
|
||
|
||
summary = self._summarize_skills(repo_info, skills)
|
||
|
||
return {
|
||
'name': summary.get('name') or repo_info.get('name', ''),
|
||
'short_desc': summary.get('short_desc') or repo_info.get('description', ''),
|
||
'description': summary.get('description', ''),
|
||
'github_url': url,
|
||
'source_repo': f"{owner}/{repo}",
|
||
'usage': summary.get('usage', ''),
|
||
'examples': summary.get('examples', ''),
|
||
'skills': skills
|
||
}
|
||
|
||
except Exception as e:
|
||
print(f"抓取 Skill 信息失败: {str(e)}")
|
||
return None
|
||
|
||
def _parse_skill_file(self, filename, content, source_url=''):
|
||
"""解析 skill 文件内容"""
|
||
try:
|
||
lines = [line.rstrip() for line in content.strip().split('\n')]
|
||
fallback_name = filename.replace('.md', '')
|
||
title = fallback_name
|
||
|
||
for line in lines:
|
||
if line.startswith('# '):
|
||
title = line[2:].strip()
|
||
break
|
||
|
||
sections = self._extract_markdown_sections(lines)
|
||
|
||
desc = self._pick_section(
|
||
sections,
|
||
['描述', '简介', 'description', 'overview', 'summary', 'what it does']
|
||
)
|
||
use_cases = self._pick_section(
|
||
sections,
|
||
['应用场景', '适用场景', '使用场景', 'use cases', 'when to use', 'use case']
|
||
)
|
||
usage = self._pick_section(
|
||
sections,
|
||
['使用方法', '用法', 'usage', 'how to use', 'quick start']
|
||
)
|
||
examples = self._pick_section(
|
||
sections,
|
||
['示例', 'examples', 'example']
|
||
)
|
||
|
||
if not desc:
|
||
desc = self._first_paragraph(lines)
|
||
|
||
content_outline = self._extract_outline(sections)
|
||
|
||
return {
|
||
'name': title,
|
||
'description': self._clean_text(desc)[:240],
|
||
'use_cases': self._clean_text(use_cases)[:240],
|
||
'usage': self._clean_text(usage)[:600],
|
||
'examples': self._clean_text(examples)[:600],
|
||
'content_outline': content_outline,
|
||
'source_url': source_url
|
||
}
|
||
except Exception:
|
||
return None
|
||
|
||
def _extract_markdown_sections(self, lines):
|
||
sections = {}
|
||
current = '__intro__'
|
||
buffer = []
|
||
|
||
for raw_line in lines:
|
||
line = raw_line.strip()
|
||
if line.startswith('##'):
|
||
sections[current] = '\n'.join(buffer).strip()
|
||
current = re.sub(r'^#+\s*', '', line).strip().lower()
|
||
buffer = []
|
||
else:
|
||
buffer.append(raw_line)
|
||
|
||
sections[current] = '\n'.join(buffer).strip()
|
||
return sections
|
||
|
||
def _pick_section(self, sections, keywords):
|
||
for key, value in sections.items():
|
||
low = key.lower()
|
||
for keyword in keywords:
|
||
if keyword.lower() in low and value.strip():
|
||
return value.strip()
|
||
return ''
|
||
|
||
def _first_paragraph(self, lines):
|
||
paragraph = []
|
||
for line in lines:
|
||
stripped = line.strip()
|
||
if not stripped or stripped.startswith('#'):
|
||
if paragraph:
|
||
break
|
||
continue
|
||
if stripped.startswith('- ') or stripped.startswith('* ') or stripped.startswith('```'):
|
||
continue
|
||
paragraph.append(stripped)
|
||
if len(' '.join(paragraph)) > 220:
|
||
break
|
||
return ' '.join(paragraph)
|
||
|
||
def _extract_outline(self, sections):
|
||
outlines = []
|
||
for key in sections.keys():
|
||
if key != '__intro__':
|
||
clean_key = key.strip()
|
||
if clean_key:
|
||
outlines.append(clean_key)
|
||
return outlines[:8]
|
||
|
||
def _clean_text(self, text):
|
||
if not text:
|
||
return ''
|
||
text = re.sub(r'```.*?```', '', text, flags=re.S)
|
||
text = re.sub(r'`([^`]+)`', r'\1', text)
|
||
text = re.sub(r'\[(.*?)\]\((.*?)\)', r'\1', text)
|
||
text = re.sub(r'^[-*]\s+', '', text, flags=re.M)
|
||
text = re.sub(r'\n{2,}', '\n', text)
|
||
text = re.sub(r'\s+', ' ', text)
|
||
return text.strip()
|
||
|
||
def _summarize_skills(self, repo_info, skills):
|
||
if not skills:
|
||
return {
|
||
'name': repo_info.get('name', ''),
|
||
'short_desc': repo_info.get('description', ''),
|
||
'description': repo_info.get('description', ''),
|
||
'usage': '',
|
||
'examples': ''
|
||
}
|
||
|
||
skill_names = [s['name'] for s in skills if s.get('name')]
|
||
use_cases = [s['use_cases'] for s in skills if s.get('use_cases')]
|
||
descs = [s['description'] for s in skills if s.get('description')]
|
||
outlines = []
|
||
for s in skills:
|
||
outlines.extend(s.get('content_outline', []))
|
||
|
||
repo_desc = (repo_info.get('description') or '').strip()
|
||
short_desc = repo_desc or (descs[0] if descs else '')
|
||
|
||
parts = []
|
||
if repo_desc:
|
||
parts.append(f"仓库简介：{repo_desc}")
|
||
if skill_names:
|
||
shown = '、'.join(skill_names[:5])
|
||
more = f" 等 {len(skill_names)} 个 skill" if len(skill_names) > 5 else ''
|
||
parts.append(f"包含内容：{shown}{more}")
|
||
if use_cases:
|
||
parts.append(f"应用场景：{use_cases[0]}")
|
||
elif descs:
|
||
parts.append(f"应用场景：{descs[0]}")
|
||
if outlines:
|
||
uniq = []
|
||
for item in outlines:
|
||
if item not in uniq:
|
||
uniq.append(item)
|
||
parts.append(f"文档涵盖：{'、'.join(uniq[:6])}")
|
||
|
||
usage_lines = []
|
||
for s in skills[:5]:
|
||
line = f"- {s['name']}"
|
||
if s.get('use_cases'):
|
||
line += f"：{s['use_cases']}"
|
||
elif s.get('description'):
|
||
line += f"：{s['description']}"
|
||
usage_lines.append(line)
|
||
|
||
examples_lines = []
|
||
for s in skills[:3]:
|
||
if s.get('source_url'):
|
||
examples_lines.append(f"{s['name']}：{s['source_url']}")
|
||
|
||
return {
|
||
'name': repo_info.get('name', ''),
|
||
'short_desc': short_desc[:180],
|
||
'description': '\n'.join(parts)[:1200],
|
||
'usage': '\n'.join(usage_lines)[:1200],
|
||
'examples': '\n'.join(examples_lines)[:1200]
|
||
}
|
||
|
||
def _extract_title(self, soup):
|
||
"""提取网站标题"""
|
||
# 优先使用 og:title
|
||
og_title = soup.find('meta', property='og:title')
|
||
if og_title and og_title.get('content'):
|
||
return og_title['content'].strip()
|
||
|
||
# 使用 title 标签
|
||
title_tag = soup.find('title')
|
||
if title_tag:
|
||
return title_tag.get_text().strip()
|
||
|
||
return ''
|
||
|
||
def _extract_description(self, soup):
|
||
"""提取网站描述"""
|
||
# 优先使用 og:description
|
||
og_desc = soup.find('meta', property='og:description')
|
||
if og_desc and og_desc.get('content'):
|
||
return og_desc['content'].strip()
|
||
|
||
# 使用 meta description
|
||
meta_desc = soup.find('meta', attrs={'name': 'description'})
|
||
if meta_desc and meta_desc.get('content'):
|
||
return meta_desc['content'].strip()
|
||
|
||
return ''
|
||
|
||
def _extract_logo(self, soup, base_url):
|
||
"""提取网站 Logo URL"""
|
||
# 优先查找 favicon
|
||
favicon = soup.find('link', rel='icon') or soup.find('link', rel='shortcut icon')
|
||
if favicon and favicon.get('href'):
|
||
return urljoin(base_url, favicon['href'])
|
||
|
||
# 查找 apple-touch-icon
|
||
apple_icon = soup.find('link', rel='apple-touch-icon')
|
||
if apple_icon and apple_icon.get('href'):
|
||
return urljoin(base_url, apple_icon['href'])
|
||
|
||
# 使用 /favicon.ico
|
||
parsed = urlparse(base_url)
|
||
return f"{parsed.scheme}://{parsed.netloc}/favicon.ico"
|
||
|
||
def download_logo(self, logo_url, save_dir='static/logos'):
|
||
"""
|
||
下载 Logo 到本地
|
||
|
||
Args:
|
||
logo_url: Logo URL
|
||
save_dir: 保存目录
|
||
|
||
Returns:
|
||
str: 保存后的文件路径，失败返回 None
|
||
"""
|
||
try:
|
||
# 确保目录存在
|
||
os.makedirs(save_dir, exist_ok=True)
|
||
|
||
# 下载图片
|
||
response = requests.get(logo_url, headers=self.headers, timeout=10, stream=True)
|
||
if response.status_code != 200:
|
||
return None
|
||
|
||
# 检查是否是图片
|
||
content_type = response.headers.get('Content-Type', '').lower()
|
||
if 'image' not in content_type:
|
||
return None
|
||
|
||
# 生成文件名
|
||
import time
|
||
import hashlib
|
||
ext = '.png'
|
||
if 'jpeg' in content_type or 'jpg' in content_type:
|
||
ext = '.jpg'
|
||
elif 'gif' in content_type:
|
||
ext = '.gif'
|
||
elif 'svg' in content_type:
|
||
ext = '.svg'
|
||
elif 'ico' in content_type:
|
||
ext = '.ico'
|
||
|
||
timestamp = str(int(time.time() * 1000))
|
||
hash_name = hashlib.md5(f"{logo_url}{timestamp}".encode()).hexdigest()[:16]
|
||
filename = f"logo_{hash_name}{ext}"
|
||
filepath = os.path.join(save_dir, filename)
|
||
|
||
# 保存图片
|
||
with open(filepath, 'wb') as f:
|
||
for chunk in response.iter_content(8192):
|
||
f.write(chunk)
|
||
|
||
# 验证图片是否有效
|
||
try:
|
||
with Image.open(filepath) as img:
|
||
img.verify()
|
||
return filepath
|
||
except:
|
||
# 图片无效，删除文件
|
||
if os.path.exists(filepath):
|
||
os.remove(filepath)
|
||
return None
|
||
|
||
except Exception as e:
|
||
print(f"下载 Logo 失败: {str(e)}")
|
||
return None |