--- name: alibabacloud-pds-multimodal-search description: | Implements exact filename search, fuzzy filename search, semantic file search, and image-based image search Triggers: "PDS drive file search", "PDS image search by image" --- # PDS Multimodal Search **Please read this entire skill document carefully** ### Features - For getting drive/drive_id, querying enterprise space, team space, personal space -> read `references/drive.md` - For uploading local files to enterprise space, team space, personal space → read `references/upload-file.md` - For downloading files from enterprise space, team space, personal space to local → read `references/download-file.md` - For searching or finding files → read `references/search-file.md` - For document/audio/video analysis, quick view, summarization on cloud drive → read `references/multianalysis-file.md` - For image search, similar image search, image-text hybrid retrieval → read `references/visual-similar-search.md` ## Agent Execution Guidelines - **Must execute steps in order**: Do not skip any step, do not proceed to the next step before the previous one is completed. - **Must follow documentation**: The aliyun pds cli commands and parameters must follow this document's guidance, do not fabricate commands. - **Recommended parameter**: All `aliyun pds` commands should include `--user-agent AlibabaCloud-Agent-Skills` parameter to help server identify request source, track usage, and troubleshoot issues. - **Must determine the target space before file operations**: Before search, upload, download, or analysis, first decide whether the user explicitly means enterprise space, team space, personal space, or all spaces. - **Space scope must not be broadened silently**: If the user explicitly says "enterprise space", only use the enterprise space drive_id. If the user explicitly says "team space", only use the matching team space drive_id. If the user explicitly says "personal space", only use the personal space drive_id. Only search across multiple spaces when the user did not restrict the scope. - **Enterprise space and team space are not interchangeable**: Even though both are returned by `list-my-group-drive`, `root_group_drive` is the enterprise space and `items` are team spaces. Never substitute one for the other. - **If the requested space is missing, stop and explain**: For example, if the user asks for enterprise space but `root_group_drive` is empty, do not fall back to a team space search. ## Core Concepts: - **Domain**: PDS instance with a unique domain_id, data is completely isolated between domains - **User**: End user under a domain, has user_id - **Group**: Team organization under a domain, divided into enterprise group and team group - **Drive**: Storage space, can belong to a user (personal space) or group (enterprise space or team space) - **File**: File or folder under a space, has file_id - **Mountapp**: PDS mount app plugin, used to mount PDS space to local, allowing users to access and manage files in PDS space conveniently ## Space Selection Rules Apply the following rules before choosing a `drive_id`: | User wording | Allowed drive source | Forbidden fallback | |------------|------|------| | "企业空间" / "company space" / "enterprise space" | `root_group_drive` only | Any drive from `items` | | "团队空间" / "某个团队空间" / "team space" | `items` only | `root_group_drive` | | "个人空间" / "我的空间" / "personal space" | `list-my-drives.items` only | group drives | | "网盘里" / "我的网盘" / no space specified | all relevant spaces | none | Before continuing, perform a brief self-check: 1. Did the user explicitly name the target space type? 2. Does the selected `drive_id` come from the correct response field for that space type? 3. If multiple team spaces exist and the user only said "team space", do I need to disambiguate which team space? If any answer is uncertain, do not guess. --- ## Installation Requirements > **Prerequisites: Requires Aliyun CLI >= 3.3.1** > > Verify CLI version: > ```bash > aliyun version # requires >= 3.3.1 > ``` > > Verify PDS plugin version: > ```bash > aliyun pds version # requires >= 0.1.4 > ``` > > If version requirements are not met, refer to `references/cli-installation-guide.md` for installation or upgrade. > > After installation, **must** enable auto plugin installation: > ```bash > aliyun configure set --auto-plugin-install true > ``` --- ## Authentication Configuration > **Prerequisites: Alibaba Cloud credentials must be configured** > > **Security Rules:** > - **Forbidden** to read, output, or print AK/SK values (e.g., `echo $ALIBABA_CLOUD_ACCESS_KEY_ID` is forbidden) > - **Forbidden** to ask users to input AK/SK directly in conversation or command line > - **Forbidden** to use `aliyun configure set` to set plaintext credentials > - **Only allowed** to use `aliyun configure list` to check credential status > > Check credential configuration: > ```bash > aliyun configure list > ``` > > Confirm the output shows a valid profile (AK, STS, or OAuth identity). > > **If no valid configuration exists, stop first.** > 1. Obtain credentials from [Alibaba Cloud Console](https://ram.console.aliyun.com/manage/ak) > 2. Configure credentials **outside this session** (run `aliyun configure` in terminal or set environment variables) > 3. Run `aliyun configure list` to verify after configuration is complete ```bash # Install Aliyun CLI (if not installed) curl -fsSL --max-time 10 https://aliyuncli.alicdn.com/install.sh | bash aliyun version # confirm >= 3.3.1 # Enable auto plugin installation aliyun configure set --auto-plugin-install true # Install Python dependencies (for multipart upload script) pip3 install requests ``` ## PDS-Specific Configuration Before executing any PDS operations, you must first configure domain_id, user_id, and authentication type -> read `references/config.md` > **Recommended parameter**: All `aliyun pds` commands should include `--user-agent AlibabaCloud-Agent-Skills` parameter > > Examples: > ```bash > aliyun pds get-user --user-agent AlibabaCloud-Agent-Skills > aliyun pds list-my-drives --user-agent AlibabaCloud-Agent-Skills > aliyun pds upload-file --drive-id --local-path --user-agent AlibabaCloud-Agent-Skills > ``` ## References | Reference Document | Path | |------------|------| | CLI Installation Guide | [references/cli-installation-guide.md](references/cli-installation-guide.md) | | RAM Permission Policies | [references/ram-policies.md](references/ram-policies.md) | ## Error Handling 1. If file search fails, please read `references/search-file.md` and strictly follow the documented process to re-execute file search. FILE:references/cli-installation-guide.md # Aliyun CLI Installation & Configuration Guide Complete guide for installing and configuring Aliyun CLI. > **Aliyun CLI 3.3.1+**: Supports installing and using all published Alibaba Cloud product plugins. Make sure to upgrade to 3.3.1 or later for full plugin ecosystem coverage. ## Installation ### macOS **Using Homebrew (Recommended)** ```bash brew install aliyun-cli # Upgrade to latest brew upgrade aliyun-cli # Verify version (>= 3.3.1) aliyun version ``` **Using Binary** ```bash # Download wget --timeout=600 https://aliyuncli.alicdn.com/aliyun-cli-macosx-latest-amd64.tgz # Extract tar -xzf aliyun-cli-macosx-latest-amd64.tgz # Move to PATH sudo mv aliyun /usr/local/bin/ # Verify aliyun version ``` ### Linux **Debian/Ubuntu** ```bash # Download wget --timeout=600 https://aliyuncli.alicdn.com/aliyun-cli-linux-latest-amd64.tgz # Extract and install tar -xzf aliyun-cli-linux-latest-amd64.tgz sudo mv aliyun /usr/local/bin/ # Verify aliyun version ``` **CentOS/RHEL** ```bash # Download wget --timeout=600 https://aliyuncli.alicdn.com/aliyun-cli-linux-latest-amd64.tgz # Extract and install tar -xzf aliyun-cli-linux-latest-amd64.tgz sudo mv aliyun /usr/local/bin/ # Verify aliyun version ``` **ARM64 Architecture** ```bash # Download ARM64 version wget --timeout=600 https://aliyuncli.alicdn.com/aliyun-cli-linux-latest-arm64.tgz # Extract and install tar -xzf aliyun-cli-linux-latest-arm64.tgz sudo mv aliyun /usr/local/bin/ ``` ### Windows **Using Binary** 1. Download from: https://aliyuncli.alicdn.com/aliyun-cli-windows-latest-amd64.zip 2. Extract the ZIP file 3. Add the directory to your PATH environment variable 4. Open new Command Prompt or PowerShell 5. Verify: `aliyun version` **Using PowerShell** ```powershell # Download Invoke-WebRequest -Uri "https://aliyuncli.alicdn.com/aliyun-cli-windows-latest-amd64.zip" -OutFile "aliyun-cli.zip" # Extract Expand-Archive -Path aliyun-cli.zip -DestinationPath C:\aliyun-cli # Add to PATH (requires admin privileges) $env:Path += ";C:\aliyun-cli" [Environment]::SetEnvironmentVariable("Path", $env:Path, [System.EnvironmentVariableTarget]::Machine) # Verify aliyun version ``` ## Configuration ### Quick Start ```bash aliyun configure set \ --mode AK \ --access-key-id \ --access-key-secret \ --region cn-hangzhou ``` All `aliyun configure` commands support non-interactive flags, which is the recommended approach — it works in scripts, CI/CD pipelines, and agent-driven automation without hanging on stdin prompts. **Where to Get Access Keys** 1. Log in to Aliyun Console: https://ram.console.aliyun.com/ 2. Navigate to: AccessKey Management 3. Create a new AccessKey pair 4. Save the secret immediately — it's only shown once ### Configuration Modes Aliyun CLI supports 6 authentication modes. All examples below use non-interactive flags. #### 1. AK Mode (Access Key) Most common mode for personal accounts and scripts. ```bash aliyun configure set \ --mode AK \ --access-key-id LTAI5tXXXXXXXX \ --access-key-secret 8dXXXXXXXXXXXXXXXXXXXXXXXX \ --region cn-hangzhou ``` Configuration is stored in `~/.aliyun/config.json`: ```json { "current": "default", "profiles": [ { "name": "default", "mode": "AK", "access_key_id": "LTAI5tXXXXXXXX", "access_key_secret": "8dXXXXXXXXXXXXXXXXXXXXXXXX", "region_id": "cn-hangzhou", "output_format": "json", "language": "en" } ] } ``` #### 2. StsToken Mode (Temporary Credentials) For short-lived access (tokens expire in 1-12 hours). ```bash aliyun configure set \ --mode StsToken \ --access-key-id LTAI5tXXXXXXXX \ --access-key-secret 8dXXXXXXXXXXXXXXXXXXXXXXXX \ --sts-token v1.0:XXXXXXXXXXXXXXXX \ --region cn-hangzhou ``` Use cases: CI/CD pipelines, temporary access for external contractors, cross-account access. #### 3. RamRoleArn Mode (Assume RAM Role) Assume a RAM role for elevated or cross-account access. ```bash aliyun configure set \ --mode RamRoleArn \ --access-key-id LTAI5tXXXXXXXX \ --access-key-secret 8dXXXXXXXXXXXXXXXXXXXXXXXX \ --ram-role-arn acs:ram::123456789012:role/AdminRole \ --role-session-name my-session \ --region cn-hangzhou ``` Use cases: cross-account resource access, temporary elevated privileges, role-based access control. #### 4. EcsRamRole Mode (ECS Instance RAM Role) Use the RAM role attached to an ECS instance — no credentials needed. ```bash aliyun configure set \ --mode EcsRamRole \ --ram-role-name MyEcsRole \ --region cn-hangzhou ``` Requirements: must be running on an ECS instance with a RAM role attached. Use cases: scripts and automation running on ECS instances. #### 5. RsaKeyPair Mode (RSA Key Pair) Use RSA key pair for authentication (generate key pair in Aliyun Console first). ```bash aliyun configure set \ --mode RsaKeyPair \ --private-key /path/to/private-key.pem \ --key-pair-name my-key-pair \ --region cn-hangzhou ``` #### 6. RamRoleArnWithEcs Mode (ECS + RAM Role) Combine ECS instance role with RAM role assumption for cross-account access from ECS. ```bash aliyun configure set \ --mode RamRoleArnWithEcs \ --ram-role-name MyEcsRole \ --ram-role-arn acs:ram::123456789012:role/TargetRole \ --role-session-name my-session \ --region cn-hangzhou ``` ### Environment Variables **Highest priority** - overrides config file **Access Key Mode** ```bash export ALIBABA_CLOUD_ACCESS_KEY_ID=your_access_key_id export ALIBABA_CLOUD_ACCESS_KEY_SECRET=your_access_key_secret export ALIBABA_CLOUD_REGION_ID=cn-hangzhou ``` **STS Token Mode** ```bash export ALIBABA_CLOUD_ACCESS_KEY_ID=your_access_key_id export ALIBABA_CLOUD_ACCESS_KEY_SECRET=your_access_key_secret export ALIBABA_CLOUD_SECURITY_TOKEN=your_sts_token export ALIBABA_CLOUD_REGION_ID=cn-hangzhou ``` **ECS RAM Role Mode** ```bash export ALIBABA_CLOUD_ECS_METADATA=role_name ``` **Use Case**: - CI/CD pipelines - Docker containers - Temporary credential override ### Managing Multiple Profiles **Create Named Profiles** ```bash aliyun configure set --profile projectA \ --mode AK \ --access-key-id LTAI5tAAAAAAAA \ --access-key-secret 8dAAAAAAAAAAAAAAAAAAAAAAAA \ --region cn-hangzhou aliyun configure set --profile projectB \ --mode AK \ --access-key-id LTAI5tBBBBBBBB \ --access-key-secret 8dBBBBBBBBBBBBBBBBBBBBBBBB \ --region cn-shanghai ``` **Use Specific Profile** ```bash aliyun ecs describe-instances --profile projectA export ALIBABA_CLOUD_PROFILE=projectA aliyun ecs describe-instances # Uses projectA ``` **List and Switch Profiles** ```bash aliyun configure list # List all profiles aliyun configure set --current projectA # Switch default profile ``` ### Credential Priority Credentials are loaded in this order (first found wins): 1. **Command-line flag**: `--profile ` 2. **Environment variable**: `ALIBABA_CLOUD_PROFILE` 3. **Environment credentials**: `ALIBABA_CLOUD_ACCESS_KEY_ID`, etc. 4. **Configuration file**: `~/.aliyun/config.json` (current profile) 5. **ECS Instance RAM Role**: If running on ECS with attached role ## Verification ### Test Authentication ```bash # Basic test - list regions aliyun ecs describe-regions # Expected output: JSON array of regions ``` **If successful**, you'll see: ```json { "Regions": { "Region": [ { "RegionId": "cn-hangzhou", "RegionEndpoint": "ecs.cn-hangzhou.aliyuncs.com", "LocalName": "华东 1（杭州）" }, ... ] }, "RequestId": "..." } ``` **If failed**, you'll see error messages: - `InvalidAccessKeyId.NotFound` - Wrong Access Key ID - `SignatureDoesNotMatch` - Wrong Access Key Secret - `InvalidSecurityToken.Expired` - STS token expired (for StsToken mode) - `Forbidden.RAM` - Insufficient permissions ### Debug Configuration ```bash # Show current configuration aliyun configure get # Test with debug logging aliyun ecs describe-regions --log-level=debug # Check credential provider aliyun configure get mode ``` ## Security Best Practices ### 1. Use RAM Users (Not Root Account) ❌ **Don't**: Use Aliyun root account credentials ✅ **Do**: Create RAM users with specific permissions ```bash # Create RAM user in console # Attach only necessary policies # Use RAM user's access keys ``` ### 2. Principle of Least Privilege Grant only the minimum permissions needed: ```bash # Example: Read-only ECS access # Attach policy: AliyunECSReadOnlyAccess ``` ### 3. Rotate Access Keys Regularly ```bash # Create new access key in RAM Console, then update configuration aliyun configure set --access-key-id NEW_KEY --access-key-secret NEW_SECRET # Delete old access key from console ``` ### 4. Use STS Tokens for Temporary Access ```bash aliyun configure set --mode StsToken \ --access-key-id XXXX --access-key-secret XXXX \ --sts-token XXXX --region cn-hangzhou ``` ### 5. Use ECS RAM Roles When Possible ```bash aliyun configure set --mode EcsRamRole --ram-role-name MyRole --region cn-hangzhou ``` ### 6. Never Commit Credentials ```bash # Add to .gitignore echo "~/.aliyun/config.json" >> .gitignore # Use environment variables in CI/CD instead ``` ### 7. Secure Config File ```bash # Restrict permissions chmod 600 ~/.aliyun/config.json ``` ## Troubleshooting ### Issue: Command Not Found ```bash # Check installation which aliyun # Check PATH echo $PATH # Reinstall or add to PATH ``` ### Issue: Authentication Failed ```bash # Verify configuration aliyun configure get # Test with debug aliyun ecs describe-regions --log-level=debug # Check credentials in console # Verify access key is active ``` ### Issue: Permission Denied ```bash # Error: Forbidden.RAM # Check RAM user permissions # Attach necessary policies in RAM console # Example: AliyunECSFullAccess for ECS operations ``` ### Issue: STS Token Expired ```bash # Error: InvalidSecurityToken.Expired # Reconfigure with new token aliyun configure set --mode StsToken \ --access-key-id XXXX --access-key-secret XXXX \ --sts-token NEW_TOKEN --region cn-hangzhou ``` ### Issue: Wrong Region ```bash # Some resources may not exist in the specified region # Check available regions aliyun ecs describe-regions # Update default region aliyun configure set region cn-shanghai ``` ## Advanced Configuration ### Custom Endpoint ```bash # Use custom or private endpoint export ALIBABA_CLOUD_ECS_ENDPOINT=ecs-vpc.cn-hangzhou.aliyuncs.com ``` ### Proxy Settings ```bash # HTTP proxy export HTTP_PROXY=http://proxy.example.com:8080 export HTTPS_PROXY=http://proxy.example.com:8080 # No proxy for specific domains export NO_PROXY=localhost,127.0.0.1,.aliyuncs.com ``` ### Timeout Settings ```bash # Connection timeout (default: 10s) export ALIBABA_CLOUD_CONNECT_TIMEOUT=30 # Read timeout (default: 10s) export ALIBABA_CLOUD_READ_TIMEOUT=30 ``` ## Next Steps After installation and configuration: 1. **Install plugins** for services you need (v3.3.1+ supports all published product plugins): ```bash aliyun plugin install --names ecs vpc rds # List all available plugins aliyun plugin list-remote ``` 2. **Explore commands**: ```bash aliyun ecs --help aliyun fc --help ``` 3. **Read documentation**: - [Command Syntax Guide](./command-syntax.md) - [Global Flags Reference](./global-flags.md) - [Common Scenarios](./common-scenarios.md) ## References - Official Documentation: https://help.aliyun.com/zh/cli/ - RAM Console: https://ram.console.aliyun.com/ - Access Key Management: https://ram.console.aliyun.com/manage/ak - Plugin Repository: https://github.com/aliyun/aliyun-cli FILE:references/config.md # PDS Aliyun CLI Configuration Guide (Important) **Scenario**: Required configuration when using aliyun pds cli for the first time **Purpose**: Configure domain_id, user_id, and authentication type for aliyun pds cli --- **Before executing any PDS operations, you must first configure domain_id, user_id, and authentication type:** ## Step 1: Verify if configuration already exists (only needs to be configured once during initialization) ```bash aliyun pds get-user --user-agent AlibabaCloud-Agent-Skills ``` If already configured successfully, it will return the current logged-in user information, and you can skip the subsequent steps. ## Step 2: Query domain list using aliyun pds list-domains (skip this step if you already have the domain_id to configure) ```bash aliyun pds list-domains --service-code edm --limit 100 --region cn-beijing --user-agent AlibabaCloud-Agent-Skills ``` The returned JSON structure is as follows. Extract the domain list from the response and display it to the user in a table format with columns `domain_id` and `domain_name`, prompting the user to select one domain. (If there is only one domain, use it directly without asking) ```json { "items": [{ "domain_id": "bj322", "domain_name": "beijing-31216", "region_id": "cn-beijing", "service_code": "edm" }], "next_marker": "" } ``` This step requires obtaining the selected domain_id before proceeding to the next step. ## Step 3: Query user list under the domain using aliyun pds list-user (skip this step if you already have the user_id to configure) ```bash # First configure domain_id with ak authentication type aliyun pds config --domain-id --authentication-type ak --user-agent AlibabaCloud-Agent-Skills # Then list users under this domain aliyun pds list-user --limit 100 --user-agent AlibabaCloud-Agent-Skills ``` The returned JSON structure is as follows. Extract the user list from the response and display it to the user in a table format with columns `user_id`, `nick_name`, `phone`, `email`, and `role`, prompting the user to select one user. (If there is only one user, use it directly without asking) ```json { "items": [ { "nick_name": "SuperAdmin", "role": "superadmin", "status": "enabled", "updated_at": 1774159173066, "phone": "123", "email": "test@example.com", "user_id": "a34527bd247e48b6b7e48d5c381b23f3" } ], "next_marker": "" } ``` This step requires obtaining the selected user_id before proceeding to the next step. ## Step 4: Configure domain_id, user_id, and authentication type to aliyun pds cli using aliyun pds config ```bash aliyun pds config \ --domain-id \ --user-id \ --authentication-type token \ --user-agent AlibabaCloud-Agent-Skills ``` **Parameter Description**: - `--domain-id`: PDS domain ID (e.g., `bj31216`), provided by PDS user, check if included in the prompt - `--user-id`: PDS user ID (e.g., `a34527bd247e48b6b7e48d5c381b23f3`), provided by PDS user, check if included in the prompt - `--authentication-type`: **Must be set to `token` if user_id parameter is provided**, indicating access with user identity **Effect After Configuration**: - No need to pass `--domain-id` parameter for subsequent PDS API calls - CLI will automatically use the configured domain_id and user_id **Verify Configuration**: ```bash # Test if configuration is effective, get-user API without parameters returns current logged-in user information in token scenario aliyun pds get-user --user-agent AlibabaCloud-Agent-Skills ``` Extract the current logged-in user information from the returned JSON: domain_id: `domain_id`, user_id: `user_id`, nick_name: `nick_name`. After successful configuration, notify the user: Current PDS DomainID: , logged-in user: () **Notes**: - Domain_id and user_id will be preset in CLI configuration - User's token will be preset in Aliyun CLI configuration file - After configuring once, no need to repeat configuration for subsequent operations --- FILE:references/download-file.md # PDS File Download Guide **Scenario**: When you have obtained the drive_id and file_id of the file to download and need to download that file **Purpose**: Download file to local --- ## Get File ID from File Path If you want to download a file from a PDS drive but only have the file path (e.g., /Photos/2026/04/vacation.jpg), you need to traverse each level of the path to find the corresponding file's file_id. The steps are as follows: For example, to download the file /Photos/2026/04/vacation.jpg from a personal space: 1. First, use the `aliyun pds list-file --drive-id --type folder --parent-file-id root --user-agent AlibabaCloud-Agent-Skills` command to list all directories under the root directory (parent-file-id=root) and find the file_id of the Photos directory: a. If the Photos directory exists, note down its file_id b. If the Photos directory does not exist, the file path is invalid 2. Use the `aliyun pds list-file --drive-id --type folder --parent-file-id --user-agent AlibabaCloud-Agent-Skills` command to list all directories under the parent directory (parent-file-id=) and find the file_id of the 2026 directory: a. If the 2026 directory exists, note down its file_id b. If the 2026 directory does not exist, the file path is invalid 3. Use the `aliyun pds list-file --drive-id --type folder --parent-file-id <2026 directory's file_id> --user-agent AlibabaCloud-Agent-Skills` command to list all directories under the parent directory (parent-file-id=<2026 directory's file_id>) and find the file_id of the 04 directory: a. If the 04 directory exists, note down its file_id b. If the 04 directory does not exist, the file path is invalid 4. Use the `aliyun pds list-file --drive-id --type file --parent-file-id <04 directory's file_id> --user-agent AlibabaCloud-Agent-Skills` command to list all files under the parent directory (parent-file-id=<04 directory's file_id>) and find the file_id of the vacation.jpg file: a. If the vacation.jpg file exists, note down its file_id b. If the vacation.jpg file does not exist, the file path is invalid 5. After obtaining the file_id of vacation.jpg, you can use this file_id to download the file **Note:** When executing the `aliyun pds list-file` command, if there are no valid items returned and the next_marker is not empty, it means that the query is not complete and the next_marker needs to be used as the --marker parameter for the next list query until next_marker is empty. --- ## Download File ### Step 1: Get Download URL Get the download link for the file: ```bash aliyun pds get-download-url \ --drive-id \ --file-id \ --expire-sec 3600 \ --user-agent AlibabaCloud-Agent-Skills ``` **Parameter Description**: - `--drive-id`: The drive_id of the space where the file is located (obtained from search results) - `--file-id`: The file_id of the file to download (obtained from search results) - `--expire-sec`: Download link validity period (seconds), default 900, maximum 115200 (32 hours) **Output**: Returns a JSON object containing `url` (download link), `expiration`, `method`, `size`, and other information. **Example Output**: ```json { "url": "https://pds-data.aliyuncs.com/...", "expiration": "2024-01-15T11:30:00Z", "method": "GET", "size": 1048576 } ``` --- ### Step 2: Download File Use the obtained download URL to download the file: ```bash curl -L --max-time 3600 --max-redirs 10 -o '' ``` **Parameter Description**: - `-L`: Follow redirects automatically (when download_URL returns a redirect URL, curl will continue downloading from the new location) - `--max-redirs 10`: Maximum number of redirects to follow (prevents infinite redirect loops) - `--max-time 3600`: Maximum time for the entire download operation (seconds) **Note**: The `-L` parameter is critical because PDS download URLs often return a redirect to the actual OSS storage URL. Without this parameter, curl will fail with a 3xx redirect response. Or use `wget`: ```bash wget --timeout=3600 --max-redirect=10 -O '' ``` **Parameter Description**: - `--max-redirect=10`: Maximum number of redirects to follow - `--timeout=3600`: Timeout for the download operation (seconds) --- ### Step 3: Verify Local File Exists FILE:references/drive.md # PDS Drive Concepts and API Reference **Scenario**: Used when querying user's drive list (including personal space, enterprise space, team space, all spaces) **Purpose**: Get drive_id for user's personal space, team space, and enterprise space --- ### Drive Concept Introduction A PDS drive is a cloud storage space that can store files. A drive must have an owner, which can be either a user or a group. - When a drive belongs to a user, it is that user's personal space. - When a drive belongs to an enterprise group, it is an enterprise space. - When a drive belongs to a team group, it is a team space. #### Users have three types of spaces in a domain: - Enterprise space - Team space - Personal space **When referring to "my PDS drive" without specifying which type of space, it should be understood as all spaces: including enterprise space, team space, and personal space** ## Mandatory Space Mapping Rules This section is critical when the user explicitly specifies a space type. - `root_group_drive` represents the enterprise space. There is at most one. - `items` returned by `list-my-group-drive` represent team spaces only. There may be multiple. - `list-my-drives.items` represent personal spaces. - Enterprise space and team space are both group-owned drives, but they are different scopes and must never be mixed. - If the user says "enterprise space", you must read `drive_id` from `root_group_drive` only. - If the user says "team space", you must read `drive_id` from `items` only. - If the user says "personal space", you must read `drive_id` from `list-my-drives.items` only. - If the user does not specify any space, then and only then can you consider all spaces together. - If the requested space does not exist in the corresponding field, stop and report that the requested space is unavailable. Do not silently switch to another space type. - If the user asks for "team space" and multiple team spaces exist, do not arbitrarily choose one unless the request already identifies which team space to use. ### Quick Decision Table | Requested scope | API field to inspect | Allowed behavior | |------|------|------| | Enterprise space | `root_group_drive` | Use it if present; otherwise stop | | Team space | `items` | Use the specified team space; ask/clarify if multiple candidates | | Personal space | `list-my-drives.items` | Use the user's personal drive | | Unspecified / all spaces | all of the above | Search one or more spaces as needed | ### Drive Query API Reference #### Query Method for Enterprise Space and Team Space You can query using the list group drives API. The items field in the response contains the user's team space list, and the root_group_drive field contains the enterprise space object. ```bash aliyun pds list-my-group-drive --limit 100 --marker "" --user-agent AlibabaCloud-Agent-Skills ``` **Output**: Returns JSON containing enterprise space and team space, including `items`, `root_group_drive`, `next_marker`, etc. Detailed explanation: - items: Contains team space list. There may be multiple team spaces. If not all displayed on one page, next_marker will be returned. If there are no team spaces, this field returns empty. - root_group_drive: Contains enterprise space object. There is at most one enterprise space. If none exists, this field returns empty. - next_marker: Used for pagination, indicates the marker for next page. Pass the returned next_marker to the marker parameter to query the next page. If no next page, this field returns empty. The JSON objects returned in items and root_group_drive are Drive objects. Important attributes of Drive objects include: - drive_id: Unique space ID, commonly used in API parameters to identify a drive (important parameter for identifying a space, other APIs may require this field as input) - drive_name: Space name, commonly used for display - total_size: Total space size in bytes - used_size: Used space size in bytes - owner_type: Owner type, either user or group - owner: Owner ID **Example Output**: ```json { "items": [ { "category": "", "created_at": "2026-03-22T06:00:12.951Z", "creator": "a34527b***c381b23f3", "description": "", "domain_id": "bj12", "drive_id": "100", "drive_name": "Test Team Space 1", "drive_type": "normal", "encrypt_data_access": false, "encrypt_mode": "none", "owner": "e71ce9***c5862d5", "owner_type": "group", "permission": null, "relative_path": "", "status": "enabled", "store_id": "fb651***943990a", "total_size": 107374182400, "updated_at": "2026-03-22T06:00:12.952Z", "used_size": 138194 }, { "category": "", "created_at": "2026-03-22T06:00:12.951Z", "creator": "a34527***81b23f3", "description": "", "domain_id": "bj12", "drive_id": "101", "drive_name": "Test Team Space 2", "drive_type": "normal", "encrypt_data_access": false, "encrypt_mode": "none", "owner": "e71ce9***b7fc5862d5", "owner_type": "group", "permission": null, "relative_path": "", "status": "enabled", "store_id": "fb6516****45c943990a", "total_size": 107374182400, "updated_at": "2026-03-22T06:00:12.952Z", "used_size": 138194 } ], "next_marker": "", "root_group_drive": { "category": "", "created_at": "2026-03-22T05:55:03.280Z", "creator": "system", "description": "", "domain_id": "bj12", "drive_id": "103", "drive_name": "Test Space", "drive_type": "normal", "encrypt_data_access": false, "encrypt_mode": "none", "owner": "9c251e****b9f952f", "owner_type": "group", "permission": null, "relative_path": "", "status": "enabled", "store_id": "fb651****43990a", "total_size": 107374182400, "updated_at": "2026-03-23T07:08:40.098Z", "used_size": 240062520 } } ``` In the above example output, team space drive_ids are: 100 and 101, enterprise space drive_id is: 103 Important interpretation rule for the example above: - If the user asked for enterprise space, only `103` is eligible. - If the user asked for team space, only `100` or `101` are eligible. - It is incorrect to use `100` or `101` as enterprise space, and incorrect to use `103` as a team space. #### Query API for Personal Space You can query using the list my drives API. The items field in the response contains the user's personal space list. ```bash aliyun pds list-my-drives --limit 100 --marker "" --user-agent AlibabaCloud-Agent-Skills ``` The JSON array in the items field returned by the personal space query API contains personal space Drive objects. Important attributes of Drive objects include: - drive_id: Unique space ID, commonly used in API parameters to identify a drive (important parameter for identifying a space, other APIs may require this field as input) - drive_name: Space name, commonly used for display - total_size: Total space size in bytes - used_size: Used space size in bytes - owner_type: Owner type, either user or group - owner: Owner ID ```json { "items": [ { "category": "", "created_at": "2026-03-22T05:59:33.037Z", "creator": "a34527b***81b23f3", "description": "", "domain_id": "bj31216", "drive_id": "108", "drive_name": "SuperAdmin (Test)", "drive_type": "normal", "encrypt_data_access": false, "encrypt_mode": "none", "owner": "a34527b***81b23f3", "owner_type": "user", "permission": null, "relative_path": "", "status": "enabled", "store_id": "fb6516***c943990a", "total_size": 107374182400, "updated_at": "2026-03-23T08:45:35.541Z", "used_size": 950709133 } ], "next_marker": "" } ``` In the above example output, personal space drive_id is 108 FILE:references/multianalysis-file.md # PDS Document and Audio/Video Analysis **Scenario**: When you have obtained the drive_id, file_id, and revision_id of the file to analyze and need to perform analysis on that file **Purpose**: Perform analysis on files and get structured analysis results --- ## Core Workflow ### Flow 1: Submit Analysis Task and Poll for Results Use Python script to automatically submit analysis task and poll until processing is complete. ```bash # Document analysis polling python scripts/pds_poll_processor.py \ --drive-id "1" \ --file-id "66e7e860a2360204b9414d5c866dd3a20af1974e" \ --revision-id "123" \ --x-pds-process "doc/analysis" \ -o doc_result.json # Audio/Video analysis polling python scripts/pds_poll_processor.py \ --drive-id "1" \ --file-id "66e7e860a2360204b9414d5c866dd3a20af1974e" \ --revision-id "123" \ --x-pds-process "video/analysis" \ -o video_result.json ``` **Parameter Description**: - `--drive-id`: The space `drive_id` where the analysis file is located - `--file-id`: The `file_id` of the file to analyze - `--revision-id`: The `revision_id` of the file to analyze - `--x-pds-process`: Processing type, `doc/analysis` (document) or `video/analysis` (audio/video). Since analysis is a synchronous API, x-pds-process must be used, not x-pds-async-process - `-o`: Save raw JSON result to file (contains signed URLs) #### Document Analysis Result Structure ```json { "summary": ["https://bucket/summary.json?sign=xxx"], "chapter_summaries": ["https://bucket/chapter_summaries.json?sign=xxx"], "keywords": ["https://bucket/keywords.json?sign=xxx"], "guiding_questions": ["https://bucket/guiding_questions.json?sign=xxx"], "method_description": ["https://bucket/method_description.json?sign=xxx"], "experiment_description": ["https://bucket/experiment_description.json?sign=xxx"], "conclusion_description": ["https://bucket/conclusion_description.json?sign=xxx"], "images": { "imgs/page_0_img_image_box_770_540_1367_860.png": { "Url": "https://bucket/imgs/page_0_img.png?sign=xxx", "Thumbnail": "https://bucket/imgs/page_0_img_thumbnail.png?sign=xxx" } } } ``` #### Audio/Video Analysis Result Structure ```json { "markdown": "https://bucket/markdown.md?sign=xxx", "summary": ["https://bucket/summary.json?sign=xxx"], "chapter_summaries": ["https://bucket/chapter_summary.json?sign=xxx"], "keywords": ["https://bucket/keywords.json?sign=xxx"], "questions": ["https://bucket/questions.json?sign=xxx"], "transcript": ["https://bucket/transcript.json?sign=xxx"], "transcript_summaries": ["https://bucket/transcript_summary.json?sign=xxx"], "transcript_chapter_summaries": ["https://bucket/transcript_chapter_summary.json?sign=xxx"], "ppt_details": ["https://bucket/ppt_details.json?sign=xxx"], "images": { "ppts/video_snapshots_0.jpg": { "Url": "https://bucket/ppts/video_snapshots_0.jpg?sign=xxx", "Thumbnail": "https://bucket/ppts/video_snapshots_0_thumbnail.jpg?sign=xxx" } } } ``` ### Flow 2: Use Formatter to Get Formatted Results Analysis results contain multiple signed URLs pointing to different types of analysis files. Use formatting scripts to parse these files and generate readable output. ```bash # Format document results python scripts/doc_analysis_formatter.py doc_result.json -o formatted_output.txt # Format audio/video results python scripts/video_analysis_formatter.py video_result.json -o formatted_output.txt ``` **Parameter Description**: - `input_file`: JSON result file path from analysis API (output from Flow 1) - `-o`: Formatted output file path (optional, outputs to console if not specified) #### Formatted Output Example The formatting script automatically downloads all files pointed to by signed URLs and generates readable output according to preset templates: ```` ================================================== 📄 【Full Summary】 ================================================== {Summary text content} 🖼️ Image: {ImagePath} (Page {PageNumber}) ================================================== 🏷️ 【Keywords】 ================================================== #{Keyword 1} | #{Keyword 2} | #{Keyword 3} | ... ================================================== 📚 【Chapter Summaries】 ================================================== ▶️ {Chapter Title} ---------------------------------------- {Chapter Content} 🖼️ Image: {ImagePath} ▶️ {Next Chapter Title} ---------------------------------------- ... ================================================== ❓ 【Guiding Questions】 ================================================== Q1: {Question 1} A1: {Answer 1} Q2: {Question 2} A2: {Answer 2} ```` Audio/video will also include dialogue transcripts and PPT extraction information. --- ### Flow 3: Extract PPT from Video If the analyzed video contains PPT, you can extract PPT from the results and generate a PPTX file. #### Prerequisites 1. Video contains PPT content 2. Analysis results contain `ppt_details` field 3. Install Python PPT processing library ```bash pip install python-pptx requests ``` #### Usage Extract PPT from video analysis results and generate PPTX file: ```bash python scripts/ppt_extraction.py video_result.json -o extracted_ppt.pptx ``` **Parameter Description**: - `input_file`: JSON result file path from video analysis API - `-o`: Output PPTX file path (default: extracted_ppt.pptx) - `--keep-aspect-ratio`: Maintain image aspect ratio (default fills entire slide) - `--validate`: Validate PPTX file after generation ##### Checklist - [ ] PPTX file can be opened with PowerPoint/WPS/LibreOffice - [ ] Slide count matches page count in `ppt_details` - [ ] Each page image is clear, no stretching or distortion - [ ] Page order matches appearance order in video - [ ] (Optional) Notes contain timestamp information ##### Auto Validation ```bash python scripts/ppt_extraction.py video_result.json --validate ``` #### Common Issues ##### 1. Feature Not Enabled ```json { "code": "OperationNotSupport", "message": "This operation is not supported." } ``` **Solution**: Contact PDS technical support to enable analysis feature. ##### 2. Signed URL Expired **Cause:** Download took too long, signed URL has expired. **Solution:** Re-request analysis results, or download all images immediately after getting results. FILE:references/ram-policies.md ## RAM Permission Requirements ### RAM Policy If minimum required permissions principle is needed: ```yaml metadata: required_permissions: - "pds:ListDomains" — List domains: list-domains - "pds:ListUser" — List users: list-user - "pds:GetDomain" — Get domain info: get-domain - "pds:ListFile" — List or search files: list-file - "pds:GetUser" — Get user info: get-user - "pds:DownloadFile" — Download file: download-file - "pds:AssumeUser" — Access via user identity token: user upload (upload-file) / download (get-download-url) / process file (file-process) / get user personal space list (list-my-drive) / get user team/enterprise space list (list-my-group-drive) / user mount (mountapp) ``` ### API and Permission Reference Table (authentication_type: token, non-RAM authentication) AssumeUser operation uses user identity access. In token scenario, except for domain management APIs and list user API, all other APIs operate after obtaining user token via AssumeUser, so the Required Permission for these operations is AssumeRole. | API Action | Required Permission | Resource | |---------------------|---------------------|------------------------------------|| | list-domains | `pds:ListDomains` | "acs:pds:*:*:domain/*", | | get-domain | `pds:GetDomain` | "acs:pds:*:*:domain/" | | list-user | `pds:ListUser` | "acs:pds:*:*:domain//*" | | search-file | `pds:AssumeUser` | "acs:pds:*:*:domain//*" | | get-user | `pds:AssumeUser` | "acs:pds:*:*:domain//*" | | get-download-url | `pds:AssumeUser` | "acs:pds:*:*:domain//*" | | process | `pds:AssumeUser` | "acs:pds:*:*:domain//*" | | list-my-group-drive | `pds:AssumeUser` | "acs:pds:*:*:domain//*" | | list-my-drives | `pds:AssumeUser` | "acs:pds:*:*:domain//*" | | upload-file | `pds:AssumeUser` | "acs:pds:*:*:domain//*" | | mountapp | `pds:AssumeUser` | "acs:pds:*:*:domain//*" | ### API and Permission Reference Table (authentication_type: ak, RAM authentication) Using ak authentication method without user identity, only the following APIs are supported: | API Action | Required Permission | Resource | |---------------------|---------------------|------------------------------------|| | list-domains | `pds:ListDomains` | "acs:pds:*:*:domain/*", | | get-domain | `pds:GetDomain` | "acs:pds:*:*:domain/" | | search-file | `pds:ListFile` | "acs:pds:*:*:domain//*" | | get-user | `pds:GetUser` | "acs:pds:*:*:domain//*" | | list-user | `pds:ListUser` | "acs:pds:*:*:domain//*" | | get-download-url | `pds:DownloadFile` | "acs:pds:*:*:domain//*" | ## Notes 1. In addition to RAM permissions, PDS also requires assigning corresponding Drive space access permissions to users in the **PDS Console** 2. When calling with AK/SK method, ensure the RAM user has the above permissions 3. When calling with Bearer Token (OAuth) method, permissions are determined by the user role within PDS FILE:references/search-file.md # PDS File Search **Scenario**: When you already have the `drive_id` to search in and need to search for files under that drive **Purpose**: Find the target files and retrieve attributes such as `file_id`. Supports scalar search based on metadata such as filename, type, size, and time, as well as multimodal semantic search based on content understanding. ## Core Workflow ### Step 1: Semantic Query Analysis Run the script `python scripts/get_semantic_query_prompt.py`, get the prompt from standard output (stdout), then use this prompt as the system prompt and the user's natural language query as user input, spawn a sub-agent to think and output JSON result, and report back to the main agent. ### Step 2: Scalar Query Analysis Run the script `python scripts/get_scalar_query_prompt.py`, get the prompt from standard output (stdout), then use this prompt as the system prompt and the user's natural language query as user input, spawn a sub-agent to think and output JSON result, and report back to the main agent. **Important**: You need to prepend current time information `UserQueryDatetime: {current time in ISO format}` to the user input, because the scalar query prompt contains time-related examples that need to reference the current time. ### Step 3: Build Query String Pass the JSON outputs from Step 1 and Step 2 to `scripts/build_query.py`: ```bash python scripts/build_query.py \ --scalar-json '{JSON output from Step 2}' \ --semantic-json '{JSON output from Step 1}' ``` The script will: 1. Recursively parse the Query object from scalar query into a query string 2. Convert semantic query to `semantic_text = "..."` format 3. Merge the modality from semantic query and the category conditions from scalar query according to the retrieval mode 4. Connect all parts with correct logical operators **Modality merge rules (important)** 1. Pure scalar retrieval supports multi-modal filtering, for example images or videos. 2. Pure semantic retrieval supports only a single modality and must converge to exactly one of `document`, `image`, `video`, or `audio`. 3. Mixed retrieval must converge to the single modality selected by semantic retrieval. - If the scalar `category` includes that semantic modality, use the semantic modality as the final modality. - If the scalar `category` conflicts with the semantic modality, do not continue the search. Instead, tell the user to adjust the conditions and try again. **Important**: If the script execution fails, it is strictly forbidden to construct `query` and `order_by` on your own understanding for the next step, as this will very easily produce syntax errors. You should go back to step one and restart the query process from the beginning. If the output `has_query` is `false`, do not execute the search, and kindly inform the user of the `message` content. If `has_query` is `true`, use the output `query` and `order_by` for the next step. ### Step 4: Execute Search Use the `query` and `order_by` output from build_query.py to call the `aliyun` CLI tool: ```bash aliyun pds search-file \ --drive-id "drive_id" \ --query "{query from build_query output}" \ --order-by "{order_by from build_query output}" \ --limit 50 \ --recursive true \ --return-total-count true \ --user-agent AlibabaCloud-Agent-Skills ``` > **Pagination**: If the response contains `next_marker`, you can pass it via `--marker` parameter in subsequent requests to get the next page. Add `--return-total-count` to get the total count of matches. ### Step 5: Display Search Results Parse the JSON output returned by the CLI tool and format the search results for display. The response structure contains an `items` array and optional `next_marker`, `total_count` fields. If there is a `next_marker`, it means there are more results available for pagination. Output error messages to stderr on failure. ## Best Practices 1. **Prefer semantic search**: When users describe file content or scenarios, semantic search is more accurate than keyword matching 2. **Combine conditions appropriately**: Semantic search can be combined with scalar conditions, for example "beach photos from this year" can use both a time range and a semantic description 3. **Distinguish pure scalar multi-modal filtering from mixed-query single-modality convergence**: - Pure scalar example: `images or videos larger than 10 MB` - Mixed, convergent example: `beach photos taken this year` - Mixed, conflicting example: `find sunset photos inside video files` 4. **Note pagination limits**: `limit` has a maximum value of 100, and large result sets require pagination 5. **Time format specification**: Time conditions use UTC format `YYYY-MM-DDTHH:mm:ss` 6. **Language consistency in semantic search**: The semantic query text must stay in the same language as the user's input. Do not translate it. FILE:references/upload-file.md # PDS File Upload Guide **Scenario**: When you have obtained the target drive_id and directory file_id and need to upload files to PDS drive **Purpose**: Upload local files to PDS drive (supports enterprise space, team space, personal space) --- ## File Upload Command Use the `aliyun pds upload-file` command to directly upload local files to PDS. This command automatically completes the three steps: create file, upload content, and complete upload. ```bash aliyun pds upload-file \ --drive-id \ --local-path \ --parent-file-id \ --name \ --check-name-mode \ --enable-rapid-upload \ --part-size \ --user-agent AlibabaCloud-Agent-Skills ``` --- ## Parameter Description | Parameter | Type | Required | Description | |------|------|------|------| | `--drive-id` | string | Yes | Target space ID (obtained from space list) | | `--local-path` | string | Yes | Full path to local file | | `--parent-file-id` | string | No | Parent directory ID, default is `root` | | `--name` | string | No | Cloud file name, defaults to local file name | | `--check-name-mode` | string | No | Name conflict handling mode: `ignore` (overwrite), `auto_rename` (auto rename), `refuse` (reject), default is `ignore` | | `--enable-rapid-upload` | bool | No | Calculate file SHA-1 for rapid upload attempt, default is `false` | | `--part-size` | int | No | Size of each part (bytes), default is 5242880 (5MB) | --- ## Common Examples ### Basic Upload Upload to root directory using local file name: ```bash aliyun pds upload-file \ --drive-id "100" \ --local-path "/path/to/file.jpg" \ --user-agent AlibabaCloud-Agent-Skills ``` ### Specify Directory and File Name Upload to specified directory with custom cloud file name: ```bash aliyun pds upload-file \ --drive-id "100" \ --local-path "/path/to/file.jpg" \ --parent-file-id "root" \ --name "my-photo.jpg" \ --check-name-mode "auto_rename" \ --user-agent AlibabaCloud-Agent-Skills ``` ### Enable Rapid Upload Calculate file SHA-1 for rapid upload attempt (completes instantly if identical file exists in cloud): ```bash aliyun pds upload-file \ --drive-id "100" \ --local-path "/path/to/file.jpg" \ --enable-rapid-upload \ --user-agent AlibabaCloud-Agent-Skills ``` ### Large File Multipart Upload Custom part size (suitable for large file uploads): ```bash aliyun pds upload-file \ --drive-id "100" \ --local-path "/path/to/large-file.zip" \ --part-size 10485760 \ --user-agent AlibabaCloud-Agent-Skills ``` ### Upload File to Specified Directory If you want to upload a file to a specified directory in a PDS drive, you need to convert the cloud directory name to the cloud directory `file_id`, and use this `file_id` as the value of the `--parent-file-id` parameter. For example, to upload a file to the /Photos/2026/04 directory in a personal space, you need to traverse each level of the /Photos/2026/04 path to find the corresponding directory's file_id. If a directory does not exist in the cloud, you need to create it. After finding the `file_id` of the final directory, use this `file_id` as the value for the --parent-file-id parameter. The steps are as follows: 1. First, use the `aliyun pds list-file --drive-id --type folder --parent-file-id root --user-agent AlibabaCloud-Agent-Skills` command to list all directories under the root directory (parent-file-id=root) and find the file_id of the Photos directory: a. If the Photos directory exists, note down its file_id b. If the Photos directory does not exist, you need to create it first and get the file_id from the creation response. Create directory command: `aliyun pds create-file --drive-id --parent-file-id root --name Photos --type folder --user-agent AlibabaCloud-Agent-Skills` 2. Use the `aliyun pds list-file --drive-id --type folder --parent-file-id --user-agent AlibabaCloud-Agent-Skills` command to list all directories under the parent directory (parent-file-id=) and find the file_id of the 2026 directory: a. If the 2026 directory exists, note down its file_id b. If the 2026 directory does not exist, you need to create it first and get the file_id from the creation response. Create directory command: `aliyun pds create-file --drive-id --parent-file-id --name 2026 --type folder --user-agent AlibabaCloud-Agent-Skills` 3. Use the `aliyun pds list-file --drive-id --type folder --parent-file-id <2026 directory's file_id> --user-agent AlibabaCloud-Agent-Skills` command to list all directories under the parent directory (parent-file-id=<2026 directory's file_id>) and find the file_id of the 04 directory: a. If the 04 directory exists, note down its file_id b. If the 04 directory does not exist, you need to create it first and get the file_id from the creation response. Create directory command: `aliyun pds create-file --drive-id --parent-file-id <2026 directory's file_id> --name 04 --type folder --user-agent AlibabaCloud-Agent-Skills` 4. After obtaining the file_id of the 04 directory, you can use this file_id as the value for the --parent-file-id parameter to upload the file to the /Photos/2026/04 directory **Note:** When executing the `aliyun pds list-file` command, if there are no valid items returned and the next_marker is not empty, it means that the query is not complete and the next_marker needs to be used as the --marker parameter for the next list query until next_marker is empty. ### Upload File to Specified Parent File ID Upload a file to a specified parent file ID, first you need to verify whether the parent directory with the specified ID exists. You can use Get File to query and verify: ```bash aliyun pds get-file \ --drive-id "100" \ --file-id "1000" \ --user-agent AlibabaCloud-Agent-Skills ``` If the specified Parent File ID does not exist, it will prompt that the parent directory does not exist and ask the user to confirm again. If this directory exists, take the response file's `parent_file_id` as the new Parent File ID and continue to query through Get File until the `parent_file_id` is `root`, indicating that the top-level directory has been found. Then concatenate the queried levels to get the full path of the file in this PDS drive space after upload. **Note:** Before uploading, you must query the full path relative to the root directory. Only after that can you proceed with the subsequent upload operations. After the query is completed, use the following command line to complete the file upload: ```bash aliyun pds upload-file \ --drive-id "100" \ --local-path "/path/to/file.jpg" \ --parent-file-id "1000" \ --user-agent AlibabaCloud-Agent-Skills ``` After the upload is completed, inform the user that the file upload was successful and display the full path relative to the root directory of the file. For example, the file has been uploaded to the `personal space`(or `team space`) and the full path is /Photos/2026/04/01/file.jpg. --- ## Output Description After successful command execution, returns a JSON object with complete file information, main fields include: - `file_id`: Unique file ID - `name`: Cloud file name - `size`: File size - `created_at`: Creation time - `updated_at`: Update time - `parent_file_id`: Parent directory ID --- ## Notes 1. **Same name file handling**: Recommend using `--check-name-mode auto_rename` to avoid overwriting existing files 2. **Rapid upload feature**: Enable `--enable-rapid-upload` to complete upload instantly when identical file exists in cloud 3. **Multipart upload**: Large files are automatically uploaded in parts, adjust part size via `--part-size` 4. **Network stability**: Ensure stable network when uploading large files to avoid interruptions FILE:references/visual-similar-search.md # Alibaba Cloud PDS Visual Similar Search Guide **Scenario**: When you have prepared a local image file or have obtained the drive_id, file_id, revision_id of an image file, and want to perform image search, similar image search, visual similarity search, or multimodal image retrieval **Purpose**: Search for similar images in the cloud drive based on user-provided image ## Step 1 [Optional]: Upload Local Image File to Drive System Space **Prerequisites** If the user has already provided the image file's drive_id, file_id, revision_id, skip this step ### Step 1.1 Get System Space Execute the following command to get the domain's system space configuration: ```bash aliyun pds get-domain --domain-id --user-agent AlibabaCloud-Agent-Skills ``` Response example: ```json { "domain_id": "bj1093", "system_drive_config": { "enable": true, "drive_id": 1, "resource_parent_file_id_map": { "value-add": "68d2348822056f5eea514146b4ad7183cdb94d2f" } } } ``` Extract the system space ID and upload parent file ID from the response. In the above example, system space ID is 1, and upload parent file ID is `68d2348822056f5eea514146b4ad7183cdb94d2f` (the `value-add` item). If `enable` in `system_drive_config` is false, or `drive_id` is empty, or `resource_parent_file_id_map` does not contain `value-add`, there is an issue with system space configuration. Please contact PDS technical support for assistance. ### Step 1.2 Upload Local File to Drive System Space Upload the local file to the drive's system space, where `drive_id` is set to the system space ID obtained in the previous step, `parent_file_id` is set to the upload parent file ID obtained in the previous step, and record the file's `file_id` and `revision_id`. ## Step 2: Construct x-pds-process If the user searches using a local file, the source file information comes from the file uploaded to the drive in Step 1.2; otherwise, the source file information comes from the drive file information provided by the user. Must call the existing Python script `scripts/render_visual_similar_search_process.py` to generate `x-pds-process`. The script will output `x-pds-process` to the terminal. **Parameter Description** - `source_domain_id`: Domain where the source image is located - `source_file_id`: File ID of the source image - `source_drive_id`: Drive ID of the source image - `source_revision_id`: Revision ID of the source image - `query`: Search semantic text, not required if none - `limit`: Maximum number of similar images to return, not required if none ```bash python scripts/render_visual_similar_search_process.py \ --source_domain_id \ --source_file_id \ --source_drive_id \ --source_revision_id \ --query \ --limit ``` --- ## Step 3: Perform Image Search **Parameter Description** - `search_drive_id`: Drive ID to search in - `search_folder_id`: File ID of the folder to search in ### Search Entire Drive ```bash aliyun pds process \ --resource-type drive \ --drive-id ${SEARCH_DRIVE_ID} \ --x-pds-process ${X_PDS_PROCESS} \ --user-agent AlibabaCloud-Agent-Skills ``` ### Search Specific Folder ```bash aliyun pds process \ --resource-type file \ --drive-id ${SEARCH_DRIVE_ID} \ --file-id ${SEARCH_FOLDER_ID} \ --x-pds-process ${X_PDS_PROCESS} \ --user-agent AlibabaCloud-Agent-Skills ``` --- ## Response **Success Response**: ```json { "similar_files": [ { "similarity": 0.95, "domain_id": "bj1093", "drive_id": "2", "file_id": "5d79206586bb5dd69fb34c349282718146c55da7", "name": "similar_image1.jpg", "type": "file", "category": "image", "size": 102400, "created_at": "2019-08-20T06:51:27.292Z", "thumbnail": "https://..." }, { "similarity": 0.84, "domain_id": "bj1093", "drive_id": "2", "file_id": "69c0e5c9432208927ca14d1f8af5e897486c6337", "name": "similar_image2.jpg", "type": "file", "category": "image", "size": 102400, "created_at": "2023-08-20T06:51:27.292Z", "thumbnail": "https://..." } ] } ``` **Field Description**: - `similarity`: Similarity score, range [0, 1], closer to 1 means more similar - `drive_id`: Drive where the result file is located (the search scope drive) - `file_id`: Similar file ID - `name`: File name - `thumbnail`: Thumbnail URL --- ## Error Handling | HTTP Status | Error Code | Description | Solution | |------------|--------|------|---------|| | 400 | InvalidParameter.xxx | Invalid parameter | Check parameter format and encoding | | 400 | OperationNotSupport | Feature not enabled | Contact PDS technical support to enable feature | | 403 | ForbiddenNoPermission.xxx | No permission | Check AccessToken permissions | **Common Errors**: ### 1. Feature Not Enabled ```json { "code": "OperationNotSupport", "message": "This operation is not supported." } ``` **Solution**: Contact PDS technical support to enable image search feature. ### 2. Insufficient Permissions ```json { "code": "ForbiddenNoPermission.file", "message": "No Permission to access resource file" } ``` **Solution**: - Ensure current user has `FILE.LIST` permission on the search space or folder - Ensure current user has `FILE.PREVIEW` permission on the source file --- ## Best Practices ### 1. Set Appropriate limit Parameter - Quick preview: `l_10` or `l_20` - Regular search: `l_50` - Comprehensive search: `l_100` (maximum) ### 2. Prefer Image-Only Retrieval Unless the user explicitly requests image-text hybrid retrieval, prefer using image-only retrieval for better accuracy --- ## FAQ **Q: Why are fewer results returned than expected?** limit only indicates the maximum number of results, it does not guarantee that limit images will be returned. A: Possible reasons: 1. Actual number of similar images is less than limit 2. Some files were filtered due to insufficient permissions 3. Total number of images in search scope is small **Q: Can I search for videos or documents?** A: Not supported, only similar image search is supported. FILE:scripts/build_query.py #!/usr/bin/env python3 # -*- coding: utf-8 -*- """ Build a PDS SearchFile API query string. Combines scalar-query JSON and semantic-query JSON into the final SearchFile API `query` string. Supports recursively parsing nested query conditions and merging `modality` with `category` constraints. Usage: python build_query.py --scalar-json '' --semantic-json '' """ import argparse import json import sys from typing import Dict, Any, Optional, List, Set, Tuple from get_scalar_query_prompt import field_schema def _escape_value(value: str) -> str: """Escape backslashes and double quotes in query values.""" return value.replace("\\", "\\\\").replace('"', '\\"') def _format_value(field: str, value: str) -> str: """ Format a value according to the field type. Args: field: Field name. value: Field value. Returns: A formatted value string. String/date fields are quoted, while long/boolean fields are not. """ # Unknown fields default to string handling. field_info = field_schema.get(field.lower(), {}) field_type = field_info.get("type", "string") # long and boolean values are emitted without quotes. if field_type in ("long", "boolean"): return str(value) # string and date values are quoted and escaped. escaped = _escape_value(str(value)) return f'"{escaped}"' def _parse_query_recursive(query: Dict[str, Any]) -> Tuple[str, Set[str]]: """ Recursively parse a Query object into a query string. Args: query: A Query JSON object. Returns: A tuple of (query_string, extracted_category_values). """ operation = query.get("Operation", "").lower() categories_found: Set[str] = set() # Operator mapping from JSON schema to SearchFile query syntax. op_map = { "lt": "<", "lte": "<=", "eq": "=", "gt": ">", "gte": ">=", "match": "match", "prefix": "prefix", } # Logical operators work on SubQueries. if operation in ("and", "or", "not"): sub_queries = query.get("SubQueries", []) if not sub_queries: return "", categories_found sub_parts = [] for sub in sub_queries: sub_str, sub_cats = _parse_query_recursive(sub) categories_found.update(sub_cats) # Filter out empty parts. This happens when category clauses are # extracted and removed from the recursive query string. if sub_str: sub_parts.append(sub_str) # Re-check the remaining subqueries after filtering. if not sub_parts: return "", categories_found if operation == "not": return f"not ({sub_parts[0]})", categories_found if len(sub_parts) == 1: return sub_parts[0], categories_found joined = f" {operation} ".join(sub_parts) return f"({joined})", categories_found # Comparison / match operators. if operation in op_map: field = query.get("Field", "") value = query.get("Value", "") api_op = op_map[operation] # Extract category constraints and rebuild them later in a dedicated # merge step instead of leaving them inline. if field.lower() == "category": categories_found.add(value) return "", categories_found formatted_value = _format_value(field, value) return f"({field} {api_op} {formatted_value})", categories_found return "", categories_found def _modality_to_category(modality: str) -> Optional[str]: """ Map a semantic modality to a scalar category. Args: modality: A modality value. Returns: The corresponding category value, or None if the modality is unsupported. """ mapping = { "document": "doc", "doc": "doc", "image": "image", "video": "video", "audio": "audio", } return mapping.get(modality.lower()) def _build_category_query(categories: Set[str]) -> str: """Build a category query fragment from a set of category values.""" if not categories: return "" if len(categories) == 1: cat = next(iter(categories)) return f'category = "{_escape_value(cat)}"' escaped_cats = [f'"{_escape_value(cat)}"' for cat in sorted(categories)] return f'category in [{", ".join(escaped_cats)}]' def build_query( scalar_json: Optional[str], semantic_json: Optional[str] ) -> Dict[str, Any]: """ Build the final query payload. Args: scalar_json: Scalar-query JSON string. semantic_json: Semantic-query JSON string. Returns: A dictionary containing has_query, query, order_by, and message. """ scalar_data = None semantic_data = None # Parse scalar query JSON. if scalar_json: try: scalar_data = json.loads(scalar_json) except json.JSONDecodeError as e: print(f"[WARN] Failed to parse scalar query JSON: {e}", file=sys.stderr) # Parse semantic query JSON. if semantic_json: try: semantic_data = json.loads(semantic_json) except json.JSONDecodeError as e: print(f"[WARN] Failed to parse semantic query JSON: {e}", file=sys.stderr) # Check whether at least one side is valid. scalar_valid = scalar_data and scalar_data.get("valid", False) semantic_valid = semantic_data and semantic_data.get("valid", False) if not scalar_valid and not semantic_valid: return { "has_query": False, "query": None, "order_by": None, "message": ( "Sorry, I can't understand your search intent yet. " "Supported search types currently include:\n" "1. File-attribute search, such as filename, type, size, and creation time\n" "2. Content-based semantic search, such as file topics or scenes\n\n" "Try describing the file more specifically, for example:\n" '- "Find last year\'s PDF documents"\n' '- "Photos of beach sunsets"\n' '- "Video files larger than 10 MB"' ), } query_parts: List[str] = [] scalar_categories: Set[str] = set() # Process scalar query. scalar_query_str = "" if scalar_valid: result = scalar_data.get("result", {}) query_obj = result.get("Query") if query_obj: # Parse recursively while extracting category clauses out of the # main scalar query string. scalar_query_str, cats_from_scalar = _parse_query_recursive(query_obj) scalar_categories.update(cats_from_scalar) # Process semantic query. semantic_query_str = "" semantic_category: Optional[str] = None if semantic_valid: result = semantic_data.get("result", {}) query_text = result.get("query", "") modalities = result.get("modality", []) if query_text: escaped_text = _escape_value(query_text) semantic_query_str = f'semantic_text = "{escaped_text}"' # Semantic search only supports a single modality. if not isinstance(modalities, list) or len(modalities) != 1: return { "has_query": False, "query": None, "order_by": None, "message": ( "Semantic search currently supports only a single modality. " "Please specify exactly one of document, image, video, or audio." ), } semantic_category = _modality_to_category(str(modalities[0])) if not semantic_category: return { "has_query": False, "query": None, "order_by": None, "message": ( "Semantic search currently supports only the four single " "modalities: document, image, video, and audio." ), } # Build the final category condition: # 1. Pure scalar retrieval allows multiple categories. # 2. Pure semantic retrieval must be single-modality. # 3. Mixed retrieval must converge to the semantic modality. final_categories: Set[str] = set() if semantic_category: if scalar_categories and semantic_category not in scalar_categories: supported_modalities = ", ".join(sorted(scalar_categories)) return { "has_query": False, "query": None, "order_by": None, "message": ( "The semantic modality conflicts with the scalar filters. " f"The semantic modality is {semantic_category}, but the " f"scalar filter only allows {supported_modalities}. " "Please adjust the conditions and try again." ), } final_categories = {semantic_category} else: final_categories = scalar_categories category_str = _build_category_query(final_categories) # Assemble the final query string. if scalar_query_str: query_parts.append(scalar_query_str) if semantic_query_str: query_parts.append(f"({semantic_query_str})") if category_str: query_parts.append(f"({category_str})") # Remove a redundant outer pair of parentheses for single-part queries. if len(query_parts) == 1: part = query_parts[0] if part.startswith("(") and part.endswith(")"): final_query = part[1:-1] else: final_query = part else: final_query = " and ".join(query_parts) # Build order_by from Sort and Order. order_by = None if scalar_valid: result = scalar_data.get("result", {}) sort_field = result.get("Sort") order_direction = result.get("Order", "") if sort_field: sort_fields = [f.strip() for f in sort_field.split(",")] order_directions = [d.strip().upper() for d in order_direction.split(",")] if order_direction else [] order_parts = [] for i, field in enumerate(sort_fields): direction = order_directions[i] if i < len(order_directions) else "ASC" if direction not in ("ASC", "DESC"): direction = "ASC" order_parts.append(f"{field} {direction}") order_by = ",".join(order_parts) return { "has_query": True, "query": final_query if final_query else None, "order_by": order_by, "message": None, } def main(): parser = argparse.ArgumentParser( description="Combine scalar-query and semantic-query JSON into a SearchFile API query string.", formatter_class=argparse.RawDescriptionHelpFormatter, epilog=""" Examples: # Scalar query only python build_query.py --scalar-json '{"valid": true, "result": {"Query": {"Operation": "gte", "Field": "size", "Value": "1000"}}}' # Semantic query only python build_query.py --semantic-json '{"valid": true, "result": {"query": "beach sunset", "modality": ["image"]}}' # Mixed query python build_query.py \\ --scalar-json '{"valid": true, "result": {"Query": {"Operation": "gt", "Field": "size", "Value": "1000"}, "Sort": "size", "Order": "desc"}}' \\ --semantic-json '{"valid": true, "result": {"query": "landscape photo", "modality": ["image"]}}' Output: { "has_query": true, "query": "combined query string", "order_by": "size DESC", "message": null } """, ) parser.add_argument( "--scalar-json", default=None, help="Scalar-query JSON string containing valid and result fields.", ) parser.add_argument( "--semantic-json", default=None, help="Semantic-query JSON string containing valid and result fields.", ) args = parser.parse_args() # Input validation. if not args.scalar_json and not args.semantic_json: print("[INFO] No query parameters were provided.", file=sys.stderr) # Build and print the final query result. result = build_query(args.scalar_json, args.semantic_json) print(json.dumps(result, ensure_ascii=False, indent=2)) if __name__ == "__main__": main() FILE:scripts/build_query_test.py #!/usr/bin/env python3 # -*- coding: utf-8 -*- """ build_query.py 全分支覆盖测试运行方式: python test_build_query.py """ import json import sys # 导入被测试的模块 from build_query import ( _escape_value, _format_value, _parse_query_recursive, _modality_to_category, build_query ) class TestResults: """测试结果统计""" def __init__(self): self.passed = 0 self.failed = 0 self.failures = [] def record(self, test_name: str, passed: bool, message: str = ""): if passed: self.passed += 1 print(f" ✓ {test_name}") else: self.failed += 1 self.failures.append((test_name, message)) print(f" ✗ {test_name}") if message: print(f" {message}") def summary(self): print("\n" + "=" * 60) print(f"测试结果: {self.passed} 通过, {self.failed} 失败") if self.failures: print("\n失败的测试:") for name, msg in self.failures: print(f" - {name}: {msg}") print("=" * 60) return self.failed == 0 results = TestResults() def test_escape_value(): """测试 _escape_value 函数""" print("\n[测试 _escape_value]") # 普通字符串 results.record( "普通字符串不变", _escape_value("hello") == "hello", f"期望 'hello', 得到 '{_escape_value('hello')}'" ) # 包含双引号 expected_quote = 'say \\"hello\\"' actual_quote = _escape_value('say "hello"') results.record( "转义双引号", actual_quote == expected_quote, f"期望 {repr(expected_quote)}, 得到 {repr(actual_quote)}" ) # 包含反斜杠 expected_slash = "path\\\\to\\\\file" actual_slash = _escape_value("path\\to\\file") results.record( "转义反斜杠", actual_slash == expected_slash, f"期望 {repr(expected_slash)}, 得到 {repr(actual_slash)}" ) # 同时包含双引号和反斜杠 expected = 'a\\\\\\"b' actual = _escape_value('a\\"b') results.record( "转义双引号和反斜杠", actual == expected, f"期望 {repr(expected)}, 得到 {repr(actual)}" ) def test_format_value(): """测试 _format_value 函数""" print("\n[测试 _format_value]") # string 类型字段 - 加引号 results.record( "string类型字段加引号 (name)", _format_value("name", "test.pdf") == '"test.pdf"', f"期望 '\"test.pdf\"', 得到 '{_format_value('name', 'test.pdf')}'" ) # long 类型字段 - 不加引号 results.record( "long类型字段不加引号 (size)", _format_value("size", "1000") == "1000", f"期望 '1000', 得到 '{_format_value('size', '1000')}'" ) # boolean 类型字段 - 不加引号 results.record( "boolean类型字段不加引号 (hidden)", _format_value("hidden", "false") == "false", f"期望 'false', 得到 '{_format_value('hidden', 'false')}'" ) results.record( "boolean类型字段不加引号 (starred)", _format_value("starred", "true") == "true", f"期望 'true', 得到 '{_format_value('starred', 'true')}'" ) # date 类型字段 - 加引号 results.record( "date类型字段加引号 (created_at)", _format_value("created_at", "2025-01-01T00:00:00") == '"2025-01-01T00:00:00"', f"期望 '\"2025-01-01T00:00:00\"', 得到 '{_format_value('created_at', '2025-01-01T00:00:00')}'" ) # 未知字段 - 默认加引号 results.record( "未知字段默认加引号", _format_value("unknown_field", "value") == '"value"', f"期望 '\"value\"', 得到 '{_format_value('unknown_field', 'value')}'" ) # string 类型带转义 expected_escaped = '"file\\"name"' actual_escaped = _format_value("name", 'file"name') results.record( "string类型带转义", actual_escaped == expected_escaped, f"期望 {repr(expected_escaped)}, 得到 {repr(actual_escaped)}" ) def test_basic_operations(): """测试基础操作符""" print("\n[测试基础操作符]") # 简单 eq 查询 - string 类型 query = {"Operation": "eq", "Field": "name", "Value": "test.pdf"} result, cats = _parse_query_recursive(query) results.record( "eq 查询 string 类型带引号", result == '(name = "test.pdf")', f"期望 '(name = \"test.pdf\")', 得到 '{result}'" ) # 简单 eq 查询 - long 类型 query = {"Operation": "eq", "Field": "size", "Value": "1000"} result, cats = _parse_query_recursive(query) results.record( "eq 查询 long 类型不带引号", result == '(size = 1000)', f"期望 '(size = 1000)', 得到 '{result}'" ) # 简单 eq 查询 - boolean 类型 query = {"Operation": "eq", "Field": "hidden", "Value": "false"} result, cats = _parse_query_recursive(query) results.record( "eq 查询 boolean 类型不带引号", result == '(hidden = false)', f"期望 '(hidden = false)', 得到 '{result}'" ) # 简单 gte 查询 - date 类型 query = {"Operation": "gte", "Field": "created_at", "Value": "2025-01-01T00:00:00"} result, cats = _parse_query_recursive(query) results.record( "gte 查询 date 类型带引号", result == '(created_at >= "2025-01-01T00:00:00")', f"期望 '(created_at >= \"2025-01-01T00:00:00\")', 得到 '{result}'" ) # match 操作符 query = {"Operation": "match", "Field": "name", "Value": "报告"} result, cats = _parse_query_recursive(query) results.record( "match 操作符", result == '(name match "报告")', f"期望 '(name match \"报告\")', 得到 '{result}'" ) # prefix 操作符 query = {"Operation": "prefix", "Field": "address", "Value": "Hang"} result, cats = _parse_query_recursive(query) results.record( "prefix 操作符", result == '(address prefix "Hang")', f"期望 '(address prefix \"Hang\")', 得到 '{result}'" ) def test_comparison_operators(): """测试比较操作符""" print("\n[测试比较操作符]") # lt - long 类型不带引号 query = {"Operation": "lt", "Field": "size", "Value": "1000"} result, cats = _parse_query_recursive(query) results.record( "lt 操作符 (long)", result == '(size < 1000)', f"期望 '(size < 1000)', 得到 '{result}'" ) # lte query = {"Operation": "lte", "Field": "size", "Value": "1000"} result, cats = _parse_query_recursive(query) results.record( "lte 操作符", result == '(size <= 1000)', f"期望 '(size <= 1000)', 得到 '{result}'" ) # gt query = {"Operation": "gt", "Field": "size", "Value": "1000"} result, cats = _parse_query_recursive(query) results.record( "gt 操作符", result == '(size > 1000)', f"期望 '(size > 1000)', 得到 '{result}'" ) # gte - date 类型带引号 query = {"Operation": "gte", "Field": "created_at", "Value": "2025-01-01T00:00:00"} result, cats = _parse_query_recursive(query) results.record( "gte 操作符 (date)", result == '(created_at >= "2025-01-01T00:00:00")', f"期望 '(created_at >= \"2025-01-01T00:00:00\")', 得到 '{result}'" ) def test_logical_operators(): """测试逻辑操作符""" print("\n[测试逻辑操作符]") # and - 多个子查询 query = { "Operation": "and", "SubQueries": [ {"Operation": "eq", "Field": "name", "Value": "test.pdf"}, {"Operation": "gt", "Field": "size", "Value": "1000"} ] } result, cats = _parse_query_recursive(query) results.record( "and 多个子查询", result == '((name = "test.pdf") and (size > 1000))', f"期望 '((name = \"test.pdf\") and (size > 1000))', 得到 '{result}'" ) # and - 单个子查询（因 category 被移除） query = { "Operation": "and", "SubQueries": [ {"Operation": "eq", "Field": "category", "Value": "image"}, {"Operation": "gt", "Field": "size", "Value": "1000"} ] } result, cats = _parse_query_recursive(query) results.record( "and 单个子查询（category被移除）", result == '(size > 1000)', f"期望 '(size > 1000)', 得到 '{result}'" ) results.record( "and 单个子查询 category 被收集", cats == {"image"}, f"期望 {{'image'}}, 得到 {cats}" ) # and - 所有子查询都是 category query = { "Operation": "and", "SubQueries": [ {"Operation": "eq", "Field": "category", "Value": "image"}, {"Operation": "eq", "Field": "category", "Value": "video"} ] } result, cats = _parse_query_recursive(query) results.record( "and 所有子查询都是category返回空", result == "", f"期望 '', 得到 '{result}'" ) results.record( "and 所有category都被收集", cats == {"image", "video"}, f"期望 {{'image', 'video'}}, 得到 {cats}" ) # or - 多个子查询 query = { "Operation": "or", "SubQueries": [ {"Operation": "eq", "Field": "name", "Value": "a.pdf"}, {"Operation": "eq", "Field": "name", "Value": "b.pdf"} ] } result, cats = _parse_query_recursive(query) results.record( "or 多个子查询", result == '((name = "a.pdf") or (name = "b.pdf"))', f"期望 '((name = \"a.pdf\") or (name = \"b.pdf\"))', 得到 '{result}'" ) # or - 单个子查询 query = { "Operation": "or", "SubQueries": [ {"Operation": "eq", "Field": "category", "Value": "doc"}, {"Operation": "eq", "Field": "name", "Value": "test.pdf"} ] } result, cats = _parse_query_recursive(query) results.record( "or 单个子查询（category被移除）", result == '(name = "test.pdf")', f"期望 '(name = \"test.pdf\")', 得到 '{result}'" ) # not 操作符 query = { "Operation": "not", "SubQueries": [ {"Operation": "eq", "Field": "hidden", "Value": "true"} ] } result, cats = _parse_query_recursive(query) results.record( "not 操作符", result == 'not ((hidden = true))', f"期望 'not ((hidden = true))', 得到 '{result}'" ) # not - 子查询为 category query = { "Operation": "not", "SubQueries": [ {"Operation": "eq", "Field": "category", "Value": "image"} ] } result, cats = _parse_query_recursive(query) results.record( "not 子查询为category返回空", result == "", f"期望 '', 得到 '{result}'" ) # 空 SubQueries query = { "Operation": "and", "SubQueries": [] } result, cats = _parse_query_recursive(query) results.record( "空SubQueries返回空", result == "", f"期望 '', 得到 '{result}'" ) def test_category_extraction(): """测试 Category 提取""" print("\n[测试 Category 提取]") # 单个 category eq query = {"Operation": "eq", "Field": "category", "Value": "image"} result, cats = _parse_query_recursive(query) results.record( "单个category eq - 返回空查询", result == "", f"期望 '', 得到 '{result}'" ) results.record( "单个category eq - 收集category", cats == {"image"}, f"期望 {{'image'}}, 得到 {cats}" ) # 多个 category 在 or 中 query = { "Operation": "or", "SubQueries": [ {"Operation": "eq", "Field": "category", "Value": "image"}, {"Operation": "eq", "Field": "category", "Value": "video"} ] } result, cats = _parse_query_recursive(query) results.record( "多个category在or中 - 返回空", result == "", f"期望 '', 得到 '{result}'" ) results.record( "多个category在or中 - 全部收集", cats == {"image", "video"}, f"期望 {{'image', 'video'}}, 得到 {cats}" ) # category 混合其他条件 query = { "Operation": "and", "SubQueries": [ {"Operation": "eq", "Field": "category", "Value": "doc"}, {"Operation": "gt", "Field": "size", "Value": "1000"} ] } result, cats = _parse_query_recursive(query) results.record( "category混合其他条件 - 只返回其他条件", result == "(size > 1000)", f"期望 '(size > 1000)', 得到 '{result}'" ) results.record( "category混合其他条件 - category被提取", cats == {"doc"}, f"期望 {{'doc'}}, 得到 {cats}" ) def test_nested_queries(): """测试嵌套查询""" print("\n[测试嵌套查询]") # 两层嵌套: and[or[A, B], C] query = { "Operation": "and", "SubQueries": [ { "Operation": "or", "SubQueries": [ {"Operation": "eq", "Field": "name", "Value": "a.pdf"}, {"Operation": "eq", "Field": "name", "Value": "b.pdf"} ] }, {"Operation": "gt", "Field": "size", "Value": "1000"} ] } result, cats = _parse_query_recursive(query) expected = '(((name = "a.pdf") or (name = "b.pdf")) and (size > 1000))' results.record( "两层嵌套 and[or[A,B], C]", result == expected, f"期望 '{expected}', 得到 '{result}'" ) # 三层嵌套 query = { "Operation": "or", "SubQueries": [ { "Operation": "and", "SubQueries": [ {"Operation": "eq", "Field": "type", "Value": "file"}, { "Operation": "or", "SubQueries": [ {"Operation": "eq", "Field": "file_extension", "Value": "pdf"}, {"Operation": "eq", "Field": "file_extension", "Value": "docx"} ] } ] }, {"Operation": "eq", "Field": "hidden", "Value": "false"} ] } result, cats = _parse_query_recursive(query) # 验证结果包含正确的结构 results.record( "三层嵌套查询", "type" in result and "file_extension" in result and "hidden" in result, f"得到 '{result}'" ) def test_semantic_queries(): """测试语义查询""" print("\n[测试语义查询]") # 纯语义查询 semantic_json = json.dumps({ "valid": True, "result": {"query": "海边日落", "modality": ["image"]} }) result = build_query(None, semantic_json) results.record( "纯语义查询", result["has_query"] == True and 'semantic_text = "海边日落"' in result["query"], f"得到 query: {result.get('query')}" ) results.record( "纯语义查询含category", "category" in result["query"], f"得到 query: {result.get('query')}" ) # 语义查询中的特殊字符转义 semantic_json = json.dumps({ "valid": True, "result": {"query": '说"你好"的照片', "modality": ["image"]} }) result = build_query(None, semantic_json) results.record( "语义查询特殊字符转义", result["has_query"] == True and '\\"' in result["query"], f"得到 query: {result.get('query')}" ) def test_category_modality_merge(): """测试 Category 和 Modality 合并""" print("\n[测试 Category/Modality 合并]") # 标量 category + 语义 modality 冲突 scalar_json = json.dumps({ "valid": True, "result": {"Query": {"Operation": "eq", "Field": "category", "Value": "doc"}} }) semantic_json = json.dumps({ "valid": True, "result": {"query": "合同", "modality": ["video"]} }) result = build_query(scalar_json, semantic_json) results.record( "category + modality 冲突时报错", result["has_query"] == False and "conflicts with the scalar filters" in result["message"], f"得到结果: has_query={result.get('has_query')}, message={result.get('message')}" ) # 标量多模态 + 语义单模态收敛到语义模态 scalar_json = json.dumps({ "valid": True, "result": { "Query": { "Operation": "or", "SubQueries": [ {"Operation": "eq", "Field": "category", "Value": "image"}, {"Operation": "eq", "Field": "category", "Value": "video"} ] } } }) semantic_json = json.dumps({ "valid": True, "result": {"query": "海边日落", "modality": ["image"]} }) result = build_query(scalar_json, semantic_json) results.record( "多模态标量与语义收敛到单模态", result["has_query"] == True and 'category = "image"' in result["query"] and "video" not in result["query"], f"得到 query: {result.get('query')}" ) # 仅语义有 modality semantic_json = json.dumps({ "valid": True, "result": {"query": "会议", "modality": ["document"]} }) result = build_query(None, semantic_json) results.record( "仅语义有modality", result["has_query"] == True and "category" in result["query"] and "doc" in result["query"], f"得到 query: {result.get('query')}" ) # 仅标量有 category scalar_json = json.dumps({ "valid": True, "result": { "Query": { "Operation": "and", "SubQueries": [ {"Operation": "eq", "Field": "category", "Value": "image"}, {"Operation": "gt", "Field": "size", "Value": "1000"} ] } } }) result = build_query(scalar_json, None) results.record( "仅标量有category", result["has_query"] == True and "category" in result["query"] and "image" in result["query"], f"得到 query: {result.get('query')}" ) def test_combined_scalar_semantic(): """测试标量+语义组合查询""" print("\n[测试标量+语义组合]") scalar_json = json.dumps({ "valid": True, "result": {"Query": {"Operation": "gt", "Field": "size", "Value": "1000"}} }) semantic_json = json.dumps({ "valid": True, "result": {"query": "风景照片", "modality": ["image"]} }) result = build_query(scalar_json, semantic_json) results.record( "标量+语义组合", result["has_query"] == True and "size > 1000" in result["query"] and "semantic_text" in result["query"], f"得到 query: {result.get('query')}" ) def test_edge_cases(): """测试边界情况""" print("\n[测试边界情况]") # 两个查询都 valid=false scalar_json = json.dumps({"valid": False}) semantic_json = json.dumps({"valid": False}) result = build_query(scalar_json, semantic_json) results.record( "两个查询都invalid", result["has_query"] == False and result["message"] is not None, f"得到 has_query: {result.get('has_query')}, message: {result.get('message')[:30]}..." ) # 只有标量 valid scalar_json = json.dumps({ "valid": True, "result": {"Query": {"Operation": "eq", "Field": "name", "Value": "test.pdf"}} }) semantic_json = json.dumps({"valid": False}) result = build_query(scalar_json, semantic_json) results.record( "只有标量valid", result["has_query"] == True and "name" in result["query"], f"得到 query: {result.get('query')}" ) # 只有语义 valid scalar_json = json.dumps({"valid": False}) semantic_json = json.dumps({ "valid": True, "result": {"query": "日落", "modality": ["image"]} }) result = build_query(scalar_json, semantic_json) results.record( "只有语义valid", result["has_query"] == True and "semantic_text" in result["query"], f"得到 query: {result.get('query')}" ) # 语义检索不允许多模态 semantic_json = json.dumps({ "valid": True, "result": {"query": "日落", "modality": ["image", "video"]} }) result = build_query(None, semantic_json) results.record( "语义检索禁止多模态", result["has_query"] == False and "only a single modality" in result["message"], f"得到结果: has_query={result.get('has_query')}, message={result.get('message')}" ) # 未知字段默认加引号 query = {"Operation": "eq", "Field": "unknown_custom_field", "Value": "test"} result_str, cats = _parse_query_recursive(query) results.record( "未知字段默认加引号", result_str == '(unknown_custom_field = "test")', f"期望 '(unknown_custom_field = \"test\")', 得到 '{result_str}'" ) # JSON 解析失败 result = build_query("invalid json", None) results.record( "JSON解析失败处理", result["has_query"] == False, f"得到 has_query: {result.get('has_query')}" ) # None 输入 result = build_query(None, None) results.record( "None输入处理", result["has_query"] == False, f"得到 has_query: {result.get('has_query')}" ) def test_sort_order(): """测试 Sort 和 Order 处理""" print("\n[测试 Sort/Order]") # 单字段排序 scalar_json = json.dumps({ "valid": True, "result": { "Query": {"Operation": "eq", "Field": "type", "Value": "file"}, "Sort": "size", "Order": "desc" } }) result = build_query(scalar_json, None) results.record( "单字段排序", result["order_by"] == "size DESC", f"期望 'size DESC', 得到 '{result.get('order_by')}'" ) # 多字段排序 scalar_json = json.dumps({ "valid": True, "result": { "Query": {"Operation": "eq", "Field": "type", "Value": "file"}, "Sort": "size,name", "Order": "desc,asc" } }) result = build_query(scalar_json, None) results.record( "多字段排序", result["order_by"] == "size DESC,name ASC", f"期望 'size DESC,name ASC', 得到 '{result.get('order_by')}'" ) # Order 数量少于 Sort（默认 ASC） scalar_json = json.dumps({ "valid": True, "result": { "Query": {"Operation": "eq", "Field": "type", "Value": "file"}, "Sort": "size,name", "Order": "desc" } }) result = build_query(scalar_json, None) results.record( "Order数量少于Sort默认ASC", result["order_by"] == "size DESC,name ASC", f"期望 'size DESC,name ASC', 得到 '{result.get('order_by')}'" ) # 无效 Order 值默认 ASC scalar_json = json.dumps({ "valid": True, "result": { "Query": {"Operation": "eq", "Field": "type", "Value": "file"}, "Sort": "size", "Order": "invalid" } }) result = build_query(scalar_json, None) results.record( "无效Order默认ASC", result["order_by"] == "size ASC", f"期望 'size ASC', 得到 '{result.get('order_by')}'" ) # 只有 Sort 没有 Order scalar_json = json.dumps({ "valid": True, "result": { "Sort": "name" } }) result = build_query(scalar_json, None) results.record( "只有Sort无Order", result["order_by"] == "name ASC", f"期望 'name ASC', 得到 '{result.get('order_by')}'" ) def test_modality_to_category(): """测试 modality 到 category 的映射""" print("\n[测试 modality 映射]") results.record( "document -> doc", _modality_to_category("document") == "doc", f"得到 '{_modality_to_category('document')}'" ) results.record( "image -> image", _modality_to_category("image") == "image", f"得到 '{_modality_to_category('image')}'" ) results.record( "video -> video", _modality_to_category("video") == "video", f"得到 '{_modality_to_category('video')}'" ) results.record( "audio -> audio", _modality_to_category("audio") == "audio", f"得到 '{_modality_to_category('audio')}'" ) results.record( "未知模态 -> None", _modality_to_category("all") is None, f"得到 '{_modality_to_category('all')}'" ) results.record( "大小写不敏感 IMAGE -> image", _modality_to_category("IMAGE") == "image", f"得到 '{_modality_to_category('IMAGE')}'" ) def test_unknown_operation(): """测试未知操作符""" print("\n[测试未知操作符]") query = {"Operation": "unknown_op", "Field": "name", "Value": "test"} result, cats = _parse_query_recursive(query) results.record( "未知操作符返回空", result == "", f"期望 '', 得到 '{result}'" ) def test_single_part_query(): """测试单一部分查询（去掉外层括号）""" print("\n[测试单一部分查询]") # 只有标量查询 scalar_json = json.dumps({ "valid": True, "result": {"Query": {"Operation": "eq", "Field": "name", "Value": "test.pdf"}} }) result = build_query(scalar_json, None) # 只有一个部分时，应该去掉外层括号 results.record( "单一标量查询去掉外层括号", result["query"] == 'name = "test.pdf"', f"期望 'name = \"test.pdf\"', 得到 '{result.get('query')}'" ) def main(): """运行所有测试""" print("=" * 60) print("build_query.py 全分支覆盖测试") print("=" * 60) # 运行所有测试 test_escape_value() test_format_value() test_basic_operations() test_comparison_operators() test_logical_operators() test_category_extraction() test_nested_queries() test_semantic_queries() test_category_modality_merge() test_combined_scalar_semantic() test_edge_cases() test_sort_order() test_modality_to_category() test_unknown_operation() test_single_part_query() # 输出结果 success = results.summary() sys.exit(0 if success else 1) if __name__ == "__main__": main() FILE:scripts/doc_analysis_formatter.py #!/usr/bin/env python3 """ PDS 文档精读结果格式化脚本功能: - 下载并解析签名文件 - 格式化输出文档分析结果 - 支持全文总结、章节总结、关键词、问题导读等注意：如需提交精读任务并轮询，请使用 pds_poll_processor.py """ import requests import json import argparse from pathlib import Path def download_and_parse(signed_url): """下载并解析签名文件""" response = requests.get(signed_url, timeout=30) response.raise_for_status() return response.json() def format_document_analysis(result, output_file=None): """ 格式化文档分析结果参数: result: 精读 API 返回的完整结果 (dict 或 JSON 文件路径) output_file: 输出文件路径，如果为 None 则打印到控制台 """ # 1. 加载结果数据 if isinstance(result, str): with open(result, 'r', encoding='utf-8') as f: result_data = json.load(f) else: result_data = result output = [] # 1. 全文总结 if "summary" in result_data and result_data["summary"]: try: summary_data = download_and_parse(result_data["summary"][0]) output.append("=" * 50) output.append("📄 【全文总结】") output.append("=" * 50) output.append("") for item in summary_data: if "Text" in item: output.append(item["Text"]) output.append("") if "Image" in item: img = item["Image"] page_num = img.get('PageNumber', 0) + 1 output.append(f"🖼️ 图片：{img['ImagePath']} (第{page_num}页)") output.append("") except Exception as e: output.append(f"⚠️ 获取全文总结失败：{e}") output.append("") # 2. 关键词 if "keywords" in result_data and result_data["keywords"]: try: keywords_data = download_and_parse(result_data["keywords"][0]) output.append("=" * 50) output.append("🏷️ 【关键词】") output.append("=" * 50) keywords_str = " | ".join([f"#{kw}" for kw in keywords_data]) output.append(keywords_str) output.append("") except Exception as e: output.append(f"⚠️ 获取关键词失败：{e}") output.append("") # 3. 章节总结 if "chapter_summaries" in result_data and result_data["chapter_summaries"]: try: chapters_data = download_and_parse(result_data["chapter_summaries"][0]) output.append("=" * 50) output.append("📚 【章节总结】") output.append("=" * 50) output.append("") for chapter in chapters_data: title = chapter.get('Title', '无标题') output.append(f"▶️ {title}") output.append("-" * 40) for item in chapter.get("Summary", []): # 兼容不同大小写的字段 text = item.get("Text") or item.get("text") if text: output.append(f" {text}") output.append("") img = item.get("Image") or item.get("image") if img: output.append(f" 🖼️ 图片：{img.get('ImagePath', '未知路径')}") output.append("") output.append("") except Exception as e: output.append(f"⚠️ 获取章节总结失败：{e}") output.append("") # 4. 问题导读 if "guiding_questions" in result_data and result_data["guiding_questions"]: try: qa_data = download_and_parse(result_data["guiding_questions"][0]) output.append("=" * 50) output.append("❓ 【问题导读】") output.append("=" * 50) output.append("") for i, qa in enumerate(qa_data, 1): output.append(f"Q{i}: {qa.get('Question', '无问题')}") output.append(f"A{i}: {qa.get('Answer', '无答案')}") output.append("") except Exception as e: output.append(f"⚠️ 获取问题导读失败：{e}") output.append("") # 5. 论文专有字段 (可选) for field_name, field_label in [ ("method_description", "方法介绍"), ("experiment_description", "实验介绍"), ("conclusion_description", "结论介绍") ]: if field_name in result_data and result_data[field_name]: try: desc_data = download_and_parse(result_data[field_name][0]) output.append("=" * 50) output.append(f"📝 【{field_label}】") output.append("=" * 50) output.append("") description = desc_data.get("Description", []) for item in description: text = item.get("text") if text: output.append(text) output.append("") img = item.get("image") if img: output.append(f"🖼️ 图片：{img.get('ImagePath', '未知路径')}") output.append("") except Exception as e: output.append(f"⚠️ 获取{field_label}失败：{e}") output.append("") # 6. 图片列表 (如果有额外图片) if "images" in result_data and result_data["images"]: output.append("=" * 50) output.append("🖼️ 【图片列表】") output.append("=" * 50) output.append("") for img_path, img_info in result_data["images"].items(): output.append(f"📎 {img_path}") if "url" in img_info: output.append(f" URL: {img_info['url']}") if "thumbnail" in img_info: output.append(f" 缩略图：{img_info['thumbnail']}") output.append("") # 输出结果 formatted_output = "\n".join(output) if output_file: with open(output_file, 'w', encoding='utf-8') as f: f.write(formatted_output) print(f"✅ 格式化结果已保存到：{output_file}") else: print(formatted_output) return formatted_output def main(): """主函数""" parser = argparse.ArgumentParser( description='PDS 文档精读结果格式化工具', formatter_class=argparse.RawDescriptionHelpFormatter, epilog=""" 示例用法: # 格式化已有的 JSON 结果文件 python doc_analysis_formatter.py result.json python doc_analysis_formatter.py result.json -o formatted_output.txt """ ) parser.add_argument( 'input_file', help='精读 API 返回的 JSON 结果文件路径' ) parser.add_argument( '-o', '--output', help='格式化输出文件路径 (默认输出到控制台)' ) args = parser.parse_args() # 检查输入文件是否存在 input_path = Path(args.input_file) if not input_path.exists(): print(f"❌ 文件不存在：{args.input_file}") return 1 try: format_document_analysis(args.input_file, args.output) return 0 except Exception as e: print(f"❌ 处理失败：{e}") import traceback traceback.print_exc() return 1 if __name__ == "__main__": exit(main()) FILE:scripts/get_scalar_query_prompt.py import json # Defines the JSON schema for metadata-based file search queries. param_schema = { "type": "object", "properties": { "Query": { "type": "object", "$id": "#query", "description": "Defines metadata search conditions for files. Nested query structures are supported.", "properties": { "Operation": { "type": "string", "enum": ["not", "or", "prefix", "and", "lt", "match", "gte", "eq", "lte", "gt"], "description": "Required. Specifies the operation type. Supported operations include:\\n- Logical operators: and, or, not (require SubQueries)\\n- Comparison operators: lt, lte, gt, gte, eq\\n- String operators: prefix, match", "examples": ["and"] }, "Field": { "type": "string", "description": "The metadata field to query. Required for all operations except logical operators (and, or, not).", "examples": ["size"] }, "Value": { "type": "string", "description": "The target value to query. All values, including numbers and timestamps, must be provided as strings. Not applicable to logical operators (and, or, not).", "examples": ["10"] }, "SubQueries": { "type": "array", "description": "Required when the operation is a logical operator (and, or, not). Contains nested query conditions that follow the logic of the parent operator. For example, when Operation is 'and', all sub-queries must be true.", "items": { "$ref": "#query" } } } }, "Sort": { "type": "string", "description": "The fields used for sorting, separated by commas. Up to 5 fields are allowed. Field order determines sort priority. For example: 'size,name'.", "examples": ["size,name"] }, "Order": { "type": "string", "enum": ["asc", "desc"], "description": "The sort direction for each field in Sort. Supported values:\\n- asc: ascending (default)\\n- desc: descending\\nYou may provide multiple directions separated by commas, for example 'asc,desc'. The number of directions cannot exceed the number of Sort fields. If a field has no explicit direction, it defaults to 'asc'.", "examples": ["asc,desc"] } }, "required": [] } # Defines supported file metadata fields and their value types. field_schema = { "parent_file_id": { "type": "string", "description": "The parent folder ID.", "examples": ["root"] }, "name": { "type": "string", "description": "The file name. Supports fuzzy matching with `match`.", "examples": ["sampleobject.jpg"] }, "type": { "type": "string", "enum": ["file", "folder"], "description": "The file type: `file` or `folder`.", "examples": ["file"] }, "file_extension": { "type": "string", "description": "The file extension without the dot, such as `pdf` or `jpg`.", "examples": ["pdf"] }, "mime_type": { "type": "string", "description": "The MIME type representing the file format.", "examples": ["image/jpeg"] }, "starred": { "type": "boolean", "description": "Whether the file is starred.", "examples": ["true"] }, "created_at": { "type": "date", "description": "The creation time in UTC, formatted as 2006-01-02T00:00:00.", "examples": ["2025-01-01T00:00:00"] }, "updated_at": { "type": "date", "description": "The last modification time in UTC, formatted as 2006-01-02T00:00:00.", "examples": ["2025-01-01T00:00:00"] }, "status": { "type": "string", "description": "The file status. Currently `available`.", "examples": ["available"] }, "hidden": { "type": "boolean", "description": "Whether the file is hidden.", "examples": ["false"] }, "size": { "type": "long", "description": "The file size in bytes.", "examples": ["1000"] }, "image_time": { "type": "date", "description": "The capture time of an image or video from EXIF metadata, formatted as 2006-01-02T00:00:00.", "examples": ["2025-01-01T00:00:00"] }, "last_access_at": { "type": "date", "description": "The most recent access time, formatted as 2006-01-02T00:00:00.", "examples": ["2025-01-01T00:00:00"] }, "category": { "type": "string", "enum": ["image", "video", "audio", "doc", "app", "others"], "description": "The file category: image, video, audio, doc, app, or others.", "examples": ["image"] }, "label": { "type": "string", "description": "The system label name.", "examples": ["landscape"] }, "face_group_id": { "type": "string", "description": "The face-group ID. Use the face-group listing API to get this ID and query photos in that group.", "examples": ["group-id-xxx"] }, "address": { "type": "string", "description": "The address. Query only one administrative level at a time, such as country ('China'), province ('Zhejiang Province'), city ('Hangzhou'), district/county ('Xihu District' or 'Tonglu County'), or street/town ('Xixi Street' or 'Sandun Town').", "examples": ["Hangzhou"] } } # Standard scalar-query JSON schema. def get_json_schema(param_schema: dict) -> dict: json_schema = { "type": "object", "properties": { "valid": { "type": "boolean", "description": "A boolean flag indicating whether the user's input can be mapped to the defined query schema. It must be false in either of these cases: 1) the input does not contain any recognizable reference to a supported field; 2) the input only contains terms for fields that are not defined in the schema, such as 'color' or 'importance'." }, "result": param_schema }, "required": ["valid"] } return json_schema # Describes how to decide whether scalar search should be used and how to extract its parameters. def schalar_search_prompt() -> str: output = f""" # Task Convert natural-language input into database query parameters: {json.dumps(param_schema, ensure_ascii=False, indent=None)} ## Supported Query Fields {json.dumps(field_schema, ensure_ascii=False, indent=None)} ## Field Validation Rules Before processing any query, you must: 1. Only process queries that clearly refer to fields defined above. 2. If the input violates this rule, for example because it does not refer to any supported field or only refers to unsupported concepts, you must return an output with `"valid": false`. Examples that must return {json.dumps({"valid": False}, ensure_ascii=False)}: - "Leshan Giant Buddha" (no field is specified; do not assume this means filename) - "red ones" (`color` is not a supported field) - "important files" (`importance` is not a supported field) Example that should not be marked invalid: - "image files" (`category eq image`) ## Query Operation Guide Each operation has a specific meaning and usage. ### Numeric comparison operations - `eq`: exact equality, such as "equals", "is", "is set to" - `gt`: greater than, such as "greater than", "over" - `gte`: greater than or equal to, such as "at least", "no less than" - `lt`: less than, such as "less than", "below" - `lte`: less than or equal to, such as "at most", "no more than" ### Text operations - `match`: search for specific text within a field - `prefix`: prefix matching for path-like values or string prefixes ### Logical operations - `and`: all conditions must be true - `or`: any condition may be true - `not`: negate a condition ## Sort and Order Guide ### Sort Supports up to 5 comma-separated sort fields. Field order determines priority. Common usage: - Single field: `"size"` - Multiple fields: `"size,name"` meaning sort by size first, then by name when sizes are equal Common mappings: - "sort by size" -> `"size"` - "sort by capture time" -> `"image_time"` - "sort by name" -> `"name"` - "sort by size and then by time" -> `"size,image_time"` Recommended combinations: - `"size,name"`: useful when sorting by size while keeping a deterministic tie-breaker - `"image_time,name"`: useful when listing results in time order while keeping a deterministic tie-breaker - `"name"`: alphabetical order ### Order Use comma-separated directions: - `asc`: ascending - `desc`: descending If fewer Order values are provided than Sort fields, the remaining fields default to `asc`. For example: - Sort: `"size,image_time,name"`, Order: `"desc"` -> equivalent to `"desc,asc,asc"` ## Important Rules 1. `match` can only be used for filename search on `name`. 2. `prefix` must not be used for `name`. 3. Follow the principle of minimal inference: only add filters that are explicitly requested by the user. For example, if the user does not mention filename conditions, the query must not contain `name` filters. This rule applies to all fields. 4. The `category` field can be used for multi-modal filtering in pure scalar search, such as images or videos. 5. If this scalar query will later be combined with semantic search, the final modality will be narrowed to the single modality chosen by semantic search. Therefore, when the user already implies a semantic target, keep `category` compatible with that semantic modality whenever possible. """.strip() output += "\n\n" output += """ ## Examples ### Example 1 Some natural-language inputs are colloquial and use abbreviations. `"The file's mime type is docx"` You should normalize the abbreviation to its canonical form whenever possible: `{"Query": {"Operation": "eq", "Field": "mime_type", "Value": "application/vnd.openxmlformats-officedocument.wordprocessingml.document"}}` `"Search for pdf files"` You should normalize the abbreviation to its canonical form whenever possible: `{"Query": {"Operation": "eq", "Field": "file_extension", "Value": "pdf"}}` ### Example 2: Time-range queries Time expressions usually imply a range rather than a single point in time. UserQueryDatetime: 2025-05-26T11:33:52+08:00 Input: `"Files created on June 6"` This should be converted into a full-day range in UTC. For the entire day of June 6 in Beijing time, the UTC start should be `2025-06-05T16:00:00` and the end should be `2025-06-06T16:00:00`. The start of day and the end boundary must always use exactly `16:00:00`. Do not use non-zero minutes or seconds at day boundaries: `{"Query": {"Operation":"and","SubQueries":[{"Operation":"gte","Field":"created_at","Value":"2025-06-05T16:00:00"},{"Operation":"lt","Field":"created_at","Value":"2025-06-06T16:00:00"}]}}` UserQueryDatetime: 2025-05-26T11:33:52+08:00 Input: `"Accessed in the last three and a half hours"` This should be converted into a UTC time span: `{"Query": {"Operation":"and","SubQueries":[{"Operation":"gte","Field":"last_access_at","Value":"2025-05-26T00:03:52"},{"Operation":"lte","Field":"last_access_at","Value":"2025-05-26T03:33:52"}]}}` Key patterns: - A calendar date = a full-day range - "recent" = a range ending at the current time - "before/after" = a bounded interval - Finally, always convert from Beijing time to UTC. The timestamp must end at whole seconds, with no milliseconds or timezone suffix. ### Example 3: Language consistency Always keep the output in the same language as the input query. Input: `"Find files whose name is 蛋糕"` Correct: `{"Query": {"Operation": "eq", "Field": "name", "Value": "蛋糕"}}` Incorrect: `{"Query": {"Operation": "eq", "Field": "name", "Value": "cake"}}` ### Example 4: Choosing the right time field There are four different time fields: - `image_time`: when an image or video was captured - Example: `"Photos taken in the summer of 2023"` -> use `image_time` - `last_access_at`: when the file was last accessed from the drive - Example: `"Files accessed yesterday"` -> use `last_access_at` - `created_at`: when the file was created or uploaded into the drive - Example: `"Files created yesterday"` -> use `created_at` - Example: `"Files uploaded yesterday"` -> use `created_at` - `updated_at`: when the file was last updated - Example: `"Files updated yesterday"` -> use `updated_at` For photo or video capture-time queries, always prefer `image_time`. ### Example 5: Basic sorting Input: `"Find files larger than 1 GB and sort by size in descending order"` `{"Query": {"Operation": "gt", "Field": "size", "Value": "1073741824"}, "Sort": "size", "Order": "desc"}` ### Example 6: Multi-field sorting Input: `"Find docx files, sort by modification time descending, then by name ascending"` `{"Query": {"Operation": "eq", "Field": "mime_type", "Value": "application/vnd.openxmlformats-officedocument.wordprocessingml.document"}, "Sort": "updated_at,name", "Order": "desc,asc"}` ### Example 7: Pure sorting with no query condition Input: `"Sort all files by name"` `{"Sort": "name"}` ### Example 8: Multi-field sorting with default order Input: `"Sort all files by size and creation time"` `{"Sort": "size,created_at"}` ### Example 9: Query simplification Input: `"Document files or text files"` `{"Query": {"Operation": "eq", "Field": "category", "Value": "doc"}}` ### Example 10: Pure scalar multi-modal filtering Input: `"Images or videos larger than 10 MB"` `{"Query": {"Operation": "and", "SubQueries": [{"Operation": "gt", "Field": "size", "Value": "10485760"}, {"Operation": "or", "SubQueries": [{"Operation": "eq", "Field": "category", "Value": "image"}, {"Operation": "eq", "Field": "category", "Value": "video"}]}]}}` """ output += "\n\n" json_schema = get_json_schema(param_schema=param_schema) output += f""" ## JSON Output Format Your output must strictly follow this JSON schema: ```json {json.dumps(json_schema, indent=None, ensure_ascii=False)} ``` ### Output Requirements - Your response must be a single valid JSON object. - Critically important: do not add any explanatory text, comments, or extra content outside the JSON object. Your entire response must contain only that JSON. """.strip() return output if __name__ == "__main__": print(schalar_search_prompt()) FILE:scripts/get_semantic_query_prompt.py def semantic_search_prompt() -> str: output = """ # Task Your task is to analyze the user's natural-language input and determine whether it contains a request for semantic (content-based) search. If it does, extract the relevant information into the specified JSON format. If the input only concerns file attributes (metadata), or is too vague to support content search, you must mark it as an invalid semantic query. ## Semantic Query Decision Guide You must decide whether the user's input meets the criteria for a semantic query. ### When to mark a query as semantic (`valid: true`) If a query describes the content, concept, or scene inside a file, it is semantic. Pay attention to patterns like these: - Content-based similarity search: "Find documents similar to this document." - Cross-modal retrieval: "Find the video corresponding to this audio clip." - Concept-level fuzzy matching: "Files about sustainable energy." - Natural-language scene descriptions: "Photos of a sunset over the sea.", "A video of someone playing guitar." Examples: - "Photos with dogs in them" (describes visual content) - "Documents about artificial intelligence" (describes a topic or concept) - "City night views" (describes a scene) ### When to mark a query as non-semantic (`valid: false`) If the query is only about file metadata or attributes, you must ignore it and mark it as invalid. This includes, but is not limited to: - Specific file attributes such as size, type, creation date, modification date, filename, or path - Exact-match or range filters on attributes - Queries that are too vague and do not describe any searchable content Examples: - "Images larger than 1 MB" (pure metadata: size) - "Files named 'report.docx'" (pure metadata: filename) - "Documents created last week" (pure metadata: date) - "Find files" (too vague) ## Mixed Queries If a query contains both semantic intent and metadata filters, for example, "Find PDF documents about artificial intelligence created this year.", your job is to extract only the semantic portion and ignore the metadata constraints. As long as a semantic portion exists, mark it as `valid: true`. - Input: "Find PDF documents about artificial intelligence created this year" - Focus on: "about artificial intelligence" -> this is the semantic portion - Ignore: "created this year", "PDF documents" -> these are metadata constraints ## Output JSON Format Your response must be a single JSON object with this structure: ```json { "type": "object", "properties": { "valid": { "type": "boolean", "description": "True if the input is a valid semantic query; otherwise false." }, "result": { "type": "object", "properties": { "query": { "type": "string", "description": "The query text to use for semantic search." }, "modality": { "type": "array", "items": { "type": "string", "enum": ["document", "image", "video", "audio"] }, "description": "The target modality. You must output exactly one modality. Do not output `all`, and do not output multiple modalities." } }, "required": ["query"], "description": "This field should exist only when 'valid' is true." } }, "required": ["valid"] } ``` - If `valid` is false, omit the `result` field. - You must output exactly one modality. - Do not output `all`, and do not output multiple modalities. - If the user's description could span multiple modalities, you must still choose the single most appropriate and executable modality. ## Semantic Query Construction Rules When constructing the `query` string inside `result`, follow these rules strictly: 1. Preserve the full context: include all relevant contextual information. Do not shorten or simplify proper nouns or specific descriptions. - Example: "Cherry blossoms at West Lake in Hangzhou" -> query: "Cherry blossoms at West Lake in Hangzhou" (not just "cherry blossoms") 2. Keep location information complete: place names must remain intact. - Example: "Architecture of the Forbidden City in Beijing" -> query: "Architecture of the Forbidden City in Beijing" 3. Preserve subject relationships: when time, place, and subject form one coherent concept, keep those relationships. - Example: "Spring scenery at West Lake in Hangzhou" -> query: "Spring scenery at West Lake in Hangzhou" 4. Language consistency is critical: the output must stay in the same language as the user's input. If the user queries in Chinese, the `query` field must be Chinese. If the user queries in English, the `query` field must be English. Do not translate the content of `query`. ## Examples ### Example 1: Pure semantic query User input: `Find some photos of beach sunsets` (Self-check: the user describes a visual scene, "beach sunsets", and explicitly specifies the modality "photos" (`image`). This is a valid semantic query.) Output: ```json { "valid": true, "result": { "query": "beach sunsets", "modality": ["image"] } } ``` ### Example 2: Pure metadata query User input: `Find video files larger than 10 MB` (Self-check: this query only refers to file attributes, specifically size, and does not describe file content. Therefore it is not a semantic query.) Output: ```json { "valid": false } ``` ### Example 3: Mixed query (semantic + metadata) User input: `Photos of family gatherings taken last summer` (Self-check: this query contains a metadata filter, "taken last summer", and a semantic concept, "family gatherings". By rule, I must ignore the metadata and extract only the semantic part.) Output: ```json { "valid": true, "result": { "query": "family gatherings", "modality": ["image"] } } ``` ### Example 4: Non-content query User input: `Find files` (Self-check: this query is too vague and does not contain any describable content for semantic search.) Output: ```json { "valid": false } ``` ### Example 5: Semantic query with a richer context User input: `Snowy scenery of the Forbidden City in Beijing` (Self-check: this query describes a scene. It does not explicitly mention a modality like "photo" or "video", but semantic retrieval must output a single modality, so I choose the most reasonable and executable one: `image`.) Output: ```json { "valid": true, "result": { "query": "Snowy scenery of the Forbidden City in Beijing", "modality": ["image"] } } ``` ### Example 6: Language consistency User input: `Find pictures of a cat sleeping on a sofa` (Self-check: the input is in English, so the `query` must also be in English. Translation is not allowed.) Correct output: ```json { "valid": true, "result": { "query": "a cat sleeping on a sofa", "modality": ["image"] } } ``` Incorrect output - do not do this: ```json { "valid": true, "result": { "query": "一只猫在沙发上睡觉", "modality": ["image"] } } ``` """.strip() return output if __name__ == "__main__": print(semantic_search_prompt()) FILE:scripts/pds_poll_processor.py #!/usr/bin/env python3 """ PDS 文档/视频精读分析轮询处理器用于自动轮询 PDS 精读分析任务，直到处理完成并下载所有结果。支持文档分析 (doc/analysis) 和视频分析 (video/analysis)。 """ import subprocess import json import time import sys from pathlib import Path from datetime import datetime class PDSPollProcessor: """PDS 精读分析轮询处理器""" def __init__(self, drive_id, file_id, revision_id, x_pds_process="doc/analysis", max_attempts=30, output_dir="/tmp"): """ 初始化处理器 Args: drive_id: 空间 ID file_id: 文件 ID revision_id: 文件版本 ID x_pds_process: 处理类型，doc/analysis 或 video/analysis max_attempts: 最大轮询次数 output_dir: 输出目录 """ self.drive_id = drive_id self.file_id = file_id self.revision_id = revision_id self.process_type = x_pds_process self.max_attempts = max_attempts self.output_dir = Path(output_dir) self.result = None # 确保输出目录存在 self.output_dir.mkdir(parents=True, exist_ok=True) def poll_analysis(self): """ 轮询精读分析结果 Returns: dict: 分析结果，如果失败则返回 None """ print("=" * 60) print(f"开始轮询 {self.process_type} 分析任务") print("=" * 60) print(f"Drive ID: {self.drive_id}") print(f"File ID: {self.file_id}") print(f"Revision ID: {self.revision_id}") print(f"Process Type: {self.process_type}") print(f"Max Attempts: {self.max_attempts}") print() # 使用列表形式构建命令，避免命令注入风险 cmd = [ "aliyun", "pds", "process", "--resource-type", "file", "--drive-id", str(self.drive_id), "--file-id", str(self.file_id), "--revision-id", str(self.revision_id), "--x-pds-process", str(self.process_type), "--user-agent", "AlibabaCloud-Agent-Skills" ] attempt = 0 while attempt < self.max_attempts: attempt += 1 timestamp = datetime.now().strftime("%H:%M:%S") print(f"[{timestamp}] ⏳ 第 {attempt}/{self.max_attempts} 次请求...") proc_result = subprocess.run( cmd, shell=False, capture_output=True, text=True, timeout=10, ) # 1. 先判断是否有错误：returncode != 0 表示 CLI 命令失败（HTTP 非 2xx） if proc_result.returncode != 0: timestamp = datetime.now().strftime("%H:%M:%S") print(f" [{timestamp}] ❌ 请求失败") print(f" 错误信息：{proc_result.stderr.strip()}") print() print("=" * 60) print("❌ 遇到错误，停止轮询") print("=" * 60) return None # 2. 解析成功响应的 Body try: response = json.loads(proc_result.stdout) # 3. 判断是否需要继续轮询（存在 retry_time 字段表示处理中） if 'retry_time' in response: retry_time = response.get('retry_time', 5) timestamp = datetime.now().strftime("%H:%M:%S") print(f" [{timestamp}] ⏰ 处理中，等待 {retry_time} 秒后重试...") time.sleep(retry_time) continue # 4. 无错误且无 retry_time，视为分析完成 timestamp = datetime.now().strftime("%H:%M:%S") print(f" [{timestamp}] ✅ 分析完成!") self.result = response return response except json.JSONDecodeError as e: timestamp = datetime.now().strftime("%H:%M:%S") print(f" [{timestamp}] ❌ JSON 解析失败：{e}") print(f" 原始输出：{proc_result.stdout[:200]}") print() print("=" * 60) print("❌ 遇到错误，停止轮询") print("=" * 60) return None except Exception as e: timestamp = datetime.now().strftime("%H:%M:%S") print(f" [{timestamp}] ❌ 发生错误：{e}") print() print("=" * 60) print("❌ 遇到错误，停止轮询") print("=" * 60) return None # 超过最大尝试次数 print() print("=" * 60) print("❌ 超过最大尝试次数，分析可能仍在进行中") print("=" * 60) return None def save_raw_result(self, filename=None): """ 保存原始 JSON 结果 (仅包含签名 URL，不下载内容) Args: filename: 文件名，默认自动生成 Returns: str: 保存的文件路径 """ if self.result is None: print("❌ 没有结果可保存") return None if filename is None: timestamp = datetime.now().strftime("%Y%m%d_%H%M%S") prefix = "doc" if self.process_type == "doc/analysis" else "video" filename = f"{prefix}_analysis_{timestamp}.json" filepath = self.output_dir / filename with open(filepath, 'w', encoding='utf-8') as f: json.dump(self.result, f, ensure_ascii=False, indent=2) print(f"💾 原始结果已保存到：{filepath}") return str(filepath) def main(): """主函数 - 命令行接口""" import argparse parser = argparse.ArgumentParser(description='PDS 文档/视频精读分析轮询处理器') parser.add_argument('--drive-id', required=True, help='空间 ID') parser.add_argument('--file-id', required=True, help='文件 ID') parser.add_argument('--revision-id', required=True, help='文件版本 ID') parser.add_argument('--x-pds-process', required=True, choices=['doc/analysis', 'video/analysis'], help='处理类型：doc/analysis（文档）或 video/analysis（音视频）') parser.add_argument('--max-attempts', type=int, default=30, help='最大轮询次数') parser.add_argument('--output-dir', default='/tmp', help='输出目录') parser.add_argument('-o', '--output', help='结果保存文件名，保存到 --output-dir 指定目录（默认 /tmp）') args = parser.parse_args() # 创建处理器 processor = PDSPollProcessor( drive_id=args.drive_id, file_id=args.file_id, revision_id=args.revision_id, x_pds_process=args.x_pds_process, max_attempts=args.max_attempts, output_dir=args.output_dir ) # 轮询分析 result = processor.poll_analysis() if result: # 保存原始结果 (包含签名 URL) processor.save_raw_result(args.output) else: print("\n❌ 分析任务失败或被中断") sys.exit(1) if __name__ == "__main__": main() FILE:scripts/ppt_extraction.py #!/usr/bin/env python3 """ PDS 视频精读 PPT 提取脚本功能: - 从视频精读结果中提取 PPT 图片 - 生成 PPTX 文件 - 支持添加备注信息 (页码、时间戳等) """ import json import requests import argparse from pathlib import Path from io import BytesIO try: from pptx import Presentation from pptx.util import Inches except ImportError: print("❌ 缺少依赖库 python-pptx") print("请运行：pip install -r scripts/requirements.txt") exit(1) def ms_to_timestamp(ms): """毫秒转时间戳格式""" seconds = ms // 1000 hours = seconds // 3600 minutes = (seconds % 3600) // 60 secs = seconds % 60 return f"{hours:02d}:{minutes:02d}:{secs:02d}" def download_image(url): """下载图片到内存""" response = requests.get(url, timeout=30) response.raise_for_status() return BytesIO(response.content) def create_pptx_from_video_analysis(result_json, output_path="output.pptx", keep_aspect_ratio=False): """ 从视频分析结果创建 PPTX 文件参数: result_json: 精读 API 返回的完整结果 (dict 或 JSON 文件路径) output_path: 输出 PPTX 文件路径 keep_aspect_ratio: 是否保持图片宽高比 (默认 False，填充整个幻灯片) 返回: bool: 成功返回 True，失败返回 False """ # 1. 加载结果数据 if isinstance(result_json, str): with open(result_json, 'r', encoding='utf-8') as f: result = json.load(f) else: result = result_json # 2. 检查 ppt_details 是否存在 if "ppt_details" not in result or not result["ppt_details"]: print("❌ 该视频中未检测到 PPT 内容") return False # 3. 下载 ppt_details JSON 文件 ppt_details_url = result["ppt_details"][0] print(f"📥 下载 PPT 详情：{ppt_details_url}") ppt_data = requests.get(ppt_details_url, timeout=30).json() # 4. 按 PPTShotIndex 排序 ppt_data.sort(key=lambda x: x["PPTShotIndex"]) print(f"📊 检测到 {len(ppt_data)} 页 PPT") # 5. 获取 images 映射 images_map = result.get("images", {}) # 6. 创建 PPTX 文件 prs = Presentation() # 设置幻灯片尺寸 (16:9 宽屏) prs.slide_width = Inches(10) prs.slide_height = Inches(5.625) # 7. 逐页添加 PPT 图片 for i, ppt_page in enumerate(ppt_data, 1): image_path = ppt_page["ImagePath"] ppt_index = ppt_page["PPTShotIndex"] start_time = ms_to_timestamp(ppt_page["StartTime"]) print(f" - 处理第 {i}/{len(ppt_data)} 页 (索引：{ppt_index}, 时间：{start_time})") # 获取图片 URL（兼容大小写） if image_path not in images_map: print(f" ⚠️ 警告：图片路径 {image_path} 未在 images 中找到，跳过") continue image_info = images_map[image_path] image_url = image_info.get("Url") or image_info.get("url") # 下载图片 try: image_stream = download_image(image_url) except Exception as e: print(f" ❌ 下载图片失败：{e}") continue # 添加空白幻灯片 blank_slide_layout = prs.slide_layouts[6] # 6 表示空白布局 slide = prs.slides.add_slide(blank_slide_layout) # 插入图片 if keep_aspect_ratio: # 保持宽高比插入 add_picture_with_aspect_ratio( slide, image_stream, prs.slide_width, prs.slide_height ) else: # 填充整个幻灯片 left = Inches(0) top = Inches(0) width = prs.slide_width height = prs.slide_height slide.shapes.add_picture( image_stream, left, top, width=width, height=height ) # 添加备注信息 notes_slide = slide.notes_slide notes_slide.notes_text_frame.text = ( f"页码：{i}\n" f"索引：{ppt_index}\n" f"出现时间：{start_time}\n" f"图片路径：{image_path}" ) # 8. 保存 PPTX 文件 prs.save(output_path) print(f"✅ PPTX 文件已生成：{output_path}") print(f" 总页数：{len(prs.slides)}") return True def add_picture_with_aspect_ratio(slide, image_stream, slide_width, slide_height): """ 插入图片并保持宽高比参数: slide: PPTX 幻灯片对象 image_stream: 图片流 (BytesIO) slide_width: 幻灯片宽度 slide_height: 幻灯片高度 """ from PIL import Image # 获取图片尺寸 img = Image.open(image_stream) img_width, img_height = img.size img_aspect = img_width / img_height slide_aspect = slide_width / slide_height if img_aspect > slide_aspect: # 图片更宽，以宽度为基准 width = slide_width height = slide_width / img_aspect left = Inches(0) top = (slide_height - height) / 2 else: # 图片更高，以高度为基准 height = slide_height width = slide_height * img_aspect top = Inches(0) left = (slide_width - width) / 2 # 重置流位置 image_stream.seek(0) slide.shapes.add_picture(image_stream, left, top, width=width, height=height) def validate_pptx(pptx_path, expected_slide_count): """ 验证生成的 PPTX 文件参数: pptx_path: PPTX 文件路径 expected_slide_count: 期望的幻灯片数量返回: bool: 验证是否通过 """ try: prs = Presentation(pptx_path) actual_count = len(prs.slides) print(f"\n📊 PPTX 验证结果:") print(f" 文件路径：{pptx_path}") print(f" 期望页数：{expected_slide_count}") print(f" 实际页数：{actual_count}") if actual_count == expected_slide_count: print(" ✅ 页数匹配") else: print(" ⚠️ 页数不匹配") # 检查每页是否包含图片 missing_images = [] for i, slide in enumerate(prs.slides, 1): has_picture = any( shape.shape_type == 13 # 13 表示图片 for shape in slide.shapes ) if not has_picture: missing_images.append(i) if missing_images: print(f" ⚠️ 以下页面未包含图片：{missing_images}") else: print(" ✅ 所有页面都包含图片") return actual_count == expected_slide_count and not missing_images except Exception as e: print(f"❌ 验证失败：{e}") return False def main(): """主函数""" parser = argparse.ArgumentParser( description='PDS 视频精读 PPT 提取工具', formatter_class=argparse.RawDescriptionHelpFormatter, epilog=""" 示例用法: python ppt_extraction.py video_analysis_result.json python ppt_extraction.py video_analysis_result.json -o extracted.pptx python ppt_extraction.py video_analysis_result.json --keep-aspect-ratio python ppt_extraction.py video_analysis_result.json --validate """ ) parser.add_argument( 'input_file', help='视频精读 API 返回的 JSON 结果文件路径' ) parser.add_argument( '-o', '--output', default='extracted_ppt.pptx', help='输出 PPTX 文件路径 (默认：extracted_ppt.pptx)' ) parser.add_argument( '--keep-aspect-ratio', action='store_true', help='保持图片宽高比 (默认填充整个幻灯片)' ) parser.add_argument( '--validate', action='store_true', help='生成后验证 PPTX 文件' ) args = parser.parse_args() # 检查输入文件是否存在 input_path = Path(args.input_file) if not input_path.exists(): print(f"❌ 文件不存在：{args.input_file}") return 1 try: success = create_pptx_from_video_analysis( args.input_file, args.output, args.keep_aspect_ratio ) if success and args.validate: # 加载 ppt_details 获取期望页数 with open(args.input_file, 'r', encoding='utf-8') as f: result = json.load(f) if "ppt_details" in result and result["ppt_details"]: ppt_details_url = result["ppt_details"][0] ppt_data = requests.get(ppt_details_url, timeout=30).json() expected_count = len(ppt_data) validate_pptx(args.output, expected_count) return 0 if success else 1 except Exception as e: print(f"❌ 处理失败：{e}") import traceback traceback.print_exc() return 1 if __name__ == "__main__": exit(main()) FILE:scripts/render_visual_similar_search_process.py import argparse import base64 from typing import Optional def url_safe_base64_encode(text): """将文本编码为 URL 安全的 base64 格式""" if not text: raise ValueError("输入文本不能为空") encoded = base64.b64encode(text.encode('utf-8')).decode('utf-8') url_safe = encoded.replace('+', '-').replace('/', '_').rstrip('=') return url_safe def generate_x_pds_process_for_vss(source_domain_id: str, source_drive_id: str, source_file_id: str, source_revision_id: str, query: Optional[str] = None, limit: Optional[int] = None) -> str: """生成以图搜图的请求参数 x-pds-process""" if not source_drive_id or not source_file_id or not source_revision_id: raise ValueError("输入参数不能为空") pds_uri = f"pds://domains/{source_domain_id}/drives/{source_drive_id}/files/{source_file_id}/revisions/{source_revision_id}" x_pds_process = f"vision/similar-search,s_{url_safe_base64_encode(pds_uri)}" if query: real_query = "semantic_text = \"{query}\"" x_pds_process += f",q_{url_safe_base64_encode(real_query)}" if limit: x_pds_process += f",l_{limit}" x_pds_process += ",/c,v_aW1hZ2U" return x_pds_process def main(): parser = argparse.ArgumentParser(description='生成以图搜图的请求参数 x-pds-process') parser.add_argument('--source_domain_id', required=True, help='要搜索的图片的 domain_id') parser.add_argument('--source_file_id', required=True, help='要搜索的图片的 file_id') parser.add_argument('--source_drive_id', required=True, help='要搜索的图片的 drive_id') parser.add_argument('--source_revision_id', required=True, help='要搜索的图片的 revision_id') parser.add_argument('--query', required=False, help='要搜索的语义文本') parser.add_argument('--limit', required=False, default=100, help='返回相似图片的最大数量') args = parser.parse_args() x_pds_process = generate_x_pds_process_for_vss(args.source_domain_id, args.source_drive_id, args.source_file_id, args.source_revision_id, args.query, args.limit) print(x_pds_process) if __name__ == '__main__': main() FILE:scripts/requirements.txt requests==2.32.2 python-pptx==0.6.23 Pillow==12.1.1 FILE:scripts/video_analysis_formatter.py #!/usr/bin/env python3 """ PDS 音视频精读结果格式化脚本功能: - 下载并解析签名文件 - 格式化输出音视频分析结果 - 支持视频总结、对话转录、章节总结、PPT 详情等注意：如需提交精读任务并轮询，请使用 pds_poll_processor.py """ import requests import json import argparse from pathlib import Path def download_and_parse(signed_url): """下载并解析签名文件""" response = requests.get(signed_url, timeout=30) response.raise_for_status() return response.json() def ms_to_timestamp(ms): """毫秒转时间戳格式""" seconds = ms // 1000 hours = seconds // 3600 minutes = (seconds % 3600) // 60 secs = seconds % 60 return f"{hours:02d}:{minutes:02d}:{secs:02d}" def format_video_analysis(result, output_file=None): """ 格式化音视频分析结果参数: result: 精读 API 返回的完整结果 (dict 或 JSON 文件路径) output_file: 输出文件路径，如果为 None 则打印到控制台 """ # 1. 加载结果数据 if isinstance(result, str): with open(result, 'r', encoding='utf-8') as f: result_data = json.load(f) else: result_data = result output = [] # 1. 视频总结 if "summary" in result_data and result_data["summary"]: try: summary_data = download_and_parse(result_data["summary"][0]) output.append("=" * 50) output.append("🎥 【视频总结】") output.append("=" * 50) output.append("") for item in summary_data: if "Text" in item: output.append(item["Text"]) output.append("") except Exception as e: output.append(f"⚠️ 获取视频总结失败：{e}") output.append("") # 2. 关键词 if "keywords" in result_data and result_data["keywords"]: try: keywords_data = download_and_parse(result_data["keywords"][0]) output.append("=" * 50) output.append("🏷️ 【关键词】") output.append("=" * 50) keywords_str = " | ".join([f"#{kw}" for kw in keywords_data]) output.append(keywords_str) output.append("") except Exception as e: output.append(f"⚠️ 获取关键词失败：{e}") output.append("") # 3. 对话转录 if "transcript" in result_data and result_data["transcript"]: try: transcript_data = download_and_parse(result_data["transcript"][0]) output.append("=" * 50) output.append("🎬 【对话转录】") output.append("=" * 50) output.append("") for item in transcript_data: start_time = ms_to_timestamp(item["TimeRange"][0]) end_time = ms_to_timestamp(item["TimeRange"][1]) speaker_id = item.get("SpeakerId", "unknown") # 提取发言人简短标识 speaker_short = speaker_id.split("-")[-1][:8] if "-" in speaker_id else speaker_id[:8] output.append(f"[{start_time} - {end_time}] 发言人 {speaker_short}:") output.append(item.get("Content", "")) output.append("") except Exception as e: output.append(f"⚠️ 获取对话转录失败：{e}") output.append("") # 4. 对话总结 if "transcript_summaries" in result_data and result_data["transcript_summaries"]: try: transcript_summary_data = download_and_parse(result_data["transcript_summaries"][0]) output.append("=" * 50) output.append("💬 【对话总结】") output.append("=" * 50) output.append("") for item in transcript_summary_data: text = item.get("Text", "") if text: output.append(text) output.append("") except Exception as e: output.append(f"⚠️ 获取对话总结失败：{e}") output.append("") # 5. 章节总结 (含时间范围) if "chapter_summaries" in result_data and result_data["chapter_summaries"]: try: chapters_data = download_and_parse(result_data["chapter_summaries"][0]) output.append("=" * 50) output.append("📚 【章节总结】") output.append("=" * 50) output.append("") for chapter in chapters_data: title = chapter.get('Title', '无标题') time_range = chapter.get('TimeRange', [0, 0]) start_time = ms_to_timestamp(time_range[0]) end_time = ms_to_timestamp(time_range[1]) output.append(f"▶️ {title} [{start_time} - {end_time}]") output.append("-" * 40) for item in chapter.get("Summary", []): text = item.get("Text") if text: output.append(f" {text}") output.append("") img = item.get("Image") if img: output.append(f" 🖼️ 图片：{img.get('ImagePath', '未知路径')}") output.append("") output.append("") except Exception as e: output.append(f"⚠️ 获取章节总结失败：{e}") output.append("") # 6. 对话章节总结 if "transcript_chapter_summaries" in result_data and result_data["transcript_chapter_summaries"]: try: transcript_chapters_data = download_and_parse(result_data["transcript_chapter_summaries"][0]) output.append("=" * 50) output.append("📖 【对话章节总结】") output.append("=" * 50) output.append("") for chapter in transcript_chapters_data: title = chapter.get('Title', '无标题') time_range = chapter.get('TimeRange', [0, 0]) start_time = ms_to_timestamp(time_range[0]) end_time = ms_to_timestamp(time_range[1]) output.append(f"▶️ {title} [{start_time} - {end_time}]") output.append("-" * 40) for item in chapter.get("Summary", []): text = item.get("Text") if text: output.append(f" {text}") output.append("") output.append("") except Exception as e: output.append(f"⚠️ 获取对话章节总结失败：{e}") output.append("") # 7. PPT 详情 if "ppt_details" in result_data and result_data["ppt_details"]: try: ppt_data = download_and_parse(result_data["ppt_details"][0]) output.append("=" * 50) output.append("📊 【PPT 提取】") output.append("=" * 50) output.append("") for i, ppt in enumerate(ppt_data, 1): page_num = ppt.get("PPTShotIndex", i - 1) + 1 start_time = ms_to_timestamp(ppt.get("StartTime", 0)) image_path = ppt.get("ImagePath", "未知路径") output.append(f"第 {page_num} 页 (出现时间：{start_time})") output.append(f"图片路径：{image_path}") output.append("") except Exception as e: output.append(f"⚠️ 获取 PPT 详情失败：{e}") output.append("") # 8. 问题导读 if "questions" in result_data and result_data["questions"]: try: qa_data = download_and_parse(result_data["questions"][0]) output.append("=" * 50) output.append("❓ 【问题导读】") output.append("=" * 50) output.append("") for i, qa in enumerate(qa_data, 1): output.append(f"Q{i}: {qa.get('Question', '无问题')}") output.append(f"A{i}: {qa.get('Answer', '无答案')}") output.append("") except Exception as e: output.append(f"⚠️ 获取问题导读失败：{e}") output.append("") # 9. 图片列表 (如果有额外图片) if "images" in result_data and result_data["images"]: output.append("=" * 50) output.append("🖼️ 【图片列表】") output.append("=" * 50) output.append("") for img_path, img_info in result_data["images"].items(): output.append(f"📎 {img_path}") if "url" in img_info: output.append(f" URL: {img_info['url']}") if "thumbnail" in img_info: output.append(f" 缩略图：{img_info['thumbnail']}") output.append("") # 输出结果 formatted_output = "\n".join(output) if output_file: with open(output_file, 'w', encoding='utf-8') as f: f.write(formatted_output) print(f"✅ 格式化结果已保存到：{output_file}") else: print(formatted_output) return formatted_output def main(): """主函数""" parser = argparse.ArgumentParser( description='PDS 音视频精读结果格式化工具', formatter_class=argparse.RawDescriptionHelpFormatter, epilog=""" 示例用法: # 格式化已有的 JSON 结果文件 python video_analysis_formatter.py result.json python video_analysis_formatter.py result.json -o formatted_output.txt """ ) parser.add_argument( 'input_file', help='精读 API 返回的 JSON 结果文件路径' ) parser.add_argument( '-o', '--output', help='格式化输出文件路径 (默认输出到控制台)' ) args = parser.parse_args() # 检查输入文件是否存在 input_path = Path(args.input_file) if not input_path.exists(): print(f"❌ 文件不存在：{args.input_file}") return 1 try: format_video_analysis(args.input_file, args.output) return 0 except Exception as e: print(f"❌ 处理失败：{e}") import traceback traceback.print_exc() return 1 if __name__ == "__main__": exit(main())

Alibabacloud Pds Multimodal Search

Comments (0)