K10小智与行空板M10 MCP服务器的院线热映电影查询

设备将自动连接到 Wi-Fi，并显示 登录 成功标志，表示已经配置网络成功，设备将在 3 秒后自动重启。

# mcp_douban_movie.py
from mcp.server.fastmcp import FastMCP
import sys
import logging
import requests
from bs4 import BeautifulSoup
import json

# 设置日志记录器
logger = logging.getLogger('DoubanMovieServer')

# 修复Windows控制台UTF-8编码问题
if sys.platform == 'win32':
    sys.stderr.reconfigure(encoding='utf-8')
    sys.stdout.reconfigure(encoding='utf-8')

# 创建MCP服务器
mcp = FastMCP("DoubanMovieServer")

@mcp.tool()
def get_now_playing_movies(city_pinyin_name: str) -> dict:
    """
    获取指定城市当前正在上映的电影列表。当需要查询当前电影院正在上映的电影时使用此工具。
    参数city_pinyin_name是城市的拼音名称，例如'beijing'、'shanghai'、'xian'等。
    返回一个字典，包含以下键：
        - 'success': 操作是否成功 (bool)
        - 'count': 实际返回的热映电影数量 (int)
        - 'result': 电影信息列表，每个元素是一个包含以下键的字典：
             * 'poster_image_url': 电影海报图片的完整URL
             * 'movie_title': 电影名称
             * 'release_date': 上映时间（年份）
             * 'directors': 导演列表
             * 'main_actors': 主演列表
             * 'rating': 电影评分（浮点数）
    """
    result = {
        'success': False,
        'count': 0,
        'result': []
    }
    
    # 构造豆瓣电影URL
    url = f'https://movie.douban.com/cinema/nowplaying/{city_pinyin_name}/'
    headers = {
        'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.124 Safari/537.36',
        'Accept-Language': 'zh-CN,zh;q=0.9,en;q=0.8'
    }
    
    try:
        logger.info(f"Fetching now playing movies for city: {city_pinyin_name}")
        
        # 发送HTTP请求
        response = requests.get(url, headers=headers, timeout=10)
        response.raise_for_status()
        
        # 解析HTML内容
        soup = BeautifulSoup(response.text, 'html.parser')
        nowplaying_div = soup.find('div', id='nowplaying')
        if not nowplaying_div:
            logger.warning("No 'nowplaying' div found on the page")
            return result
            
        movie_items = nowplaying_div.select('ul.lists > li.list-item')
        
        if not movie_items:
            logger.warning("No movie items found in the 'nowplaying' section")
            return result
        
        # 提取电影信息
        movies = []
        for item in movie_items:
            movie = extract_movie_info(item)
            if movie:
                movies.append(movie)
        
        # 构建结果
        result['success'] = True
        result['count'] = len(movies)
        result['result'] = movies
        logger.info(f"Successfully fetched {len(movies)} movies for {city_pinyin_name}")
    
    except requests.exceptions.RequestException as e:
        logger.error(f"Request failed: {e}")
    except Exception as e:
        logger.error(f"Error occurred: {e}")
    
    return result

def extract_movie_info(item):
    try:
        # 从data属性获取基本信息
        data_title = item.get('data-title', '')
        data_director = item.get('data-director', '')
        data_actors = item.get('data-actors', '')
        data_release = item.get('data-release', '')
        data_score = item.get('data-score', '0')
        
        # 处理导演和演员信息
        directors = [d.strip() for d in data_director.split('/')] if data_director else []
        main_actors = [a.strip() for a in data_actors.split('/')] if data_actors else []
        
        # 海报图片URL
        poster_img = item.select_one('li.poster img')
        poster_image_url = poster_img['src'] if poster_img and 'src' in poster_img.attrs else ""
        
        # 电影标题 - 优先使用data-title属性
        movie_title = data_title
        
        # 评分处理
        rating = 0.0
        try:
            # 尝试从data-score属性获取评分
            if data_score and data_score != '0':
                rating = float(data_score)
            else:
                # 如果data-score不可用，尝试从页面元素获取
                rating_tag = item.select_one('li.srating span.subject-rate')
                if rating_tag and rating_tag.text.strip():
                    rating = float(rating_tag.text.strip())
        except (ValueError, TypeError):
            pass
        
        movie_info = {
            'poster_image_url': poster_image_url,
            'movie_title': movie_title,
            'release_date': data_release,
            'directors': directors,
            'main_actors': main_actors,
            'rating': rating
        }
        
        logger.debug(f"Extracted movie: {movie_info['movie_title']}")
        return movie_info
    
    except Exception as e:
        logger.error(f"Error extracting movie info: {e}")
        return None

# 启动服务器
if __name__ == "__main__":
    mcp.run(transport="stdio")
复制代码