--- name: alibabacloud-pds-intelligent-workspace description: | Implements file upload, file download, document analysis, video analysis, and image editing features. Image editing supports scaling, cropping, rotation, segmentation, removal, watermark, and other operations with save-as to PDS. Access cloud drive storage via mount app. The mount app installation process involves driver installation and creating scheduled tasks/launchd. Triggers: "upload file to PDS drive", "download file from PDS drive", "PDS drive document analysis", "PDS drive video analysis", "PDS image editing", "PDS image processing", "mount PDS drive", "install mount app", "uninstall mount app", "PDS drive mount access", "stop mount app" --- # PDS (Cloud Drive) **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` - For mount app, install mount app, uninstall mount app, stop mount app → read `references/mountapp.md` - For image editing, image processing → read `references/image-editing.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. ## 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 team (team/enterprise 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 --- ## 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/chat.md 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** ### 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 #### 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/image-editing.md # PDS Image Editing Guide **Scenario**: Already obtained drive_id, file_id, revision_id, need to perform image editing operations **Purpose**: Edit images through the Process interface, including scaling, cropping, rotation, segmentation, removal, watermark and other features, and save the results to PDS --- ## Image Editing Capabilities Overview PDS image editing capabilities are implemented through `x-pds-process=image/xxx`, supporting basic image processing such as scaling, cropping, rotation, as well as AI image processing such as segmentation and removal. | Parameter | Description | Reference Link | |------|------|----------------------------------------------------------------------------------------------------------------------------------------| | resize | Scale image to specified size | [resize documentation](https://help.aliyun.com/zh/oss/user-guide/resize-images-4) | | watermark | Add text or image watermark to image | [watermark documentation](https://help.aliyun.com/zh/oss/user-guide/add-watermarks) | | crop | Crop rectangular image of specified size | [crop documentation](https://help.aliyun.com/zh/oss/user-guide/custom-crop) | | quality | Adjust quality of JPEG and WebP format images | [quality documentation](https://help.aliyun.com/zh/oss/user-guide/adjust-image-quality) | | format | Convert image format | [format documentation](https://help.aliyun.com/zh/oss/user-guide/convert-image-formats-2) | | auto-orient | Auto-rotate images with rotation parameters | [auto-orient documentation](https://help.aliyun.com/zh/oss/user-guide/auto-rotate-4) | | circle | Crop circular image with specified size centered on image | [circle documentation](https://help.aliyun.com/zh/oss/user-guide/circle-crop) | | indexcrop | Slice image by position on x or y axis, then select one image | [indexcrop documentation](https://help.aliyun.com/zh/oss/user-guide/indexed-slice) | | rounded-corners | Crop image into rounded rectangle with specified corner radius | [rounded-corners documentation](https://help.aliyun.com/zh/oss/user-guide/rounded-rectangle-4) | | blur | Apply blur effect to image | [blur documentation](https://help.aliyun.com/zh/oss/user-guide/blur) | | rotate | Rotate image clockwise by specified angle | [rotate documentation](https://help.aliyun.com/zh/oss/user-guide/rotate) | | interlace | Adjust JPG images to progressive display | [interlace documentation](https://help.aliyun.com/zh/oss/user-guide/gradual-display) | | bright | Adjust image brightness | [bright documentation](https://help.aliyun.com/zh/oss/user-guide/brightness) | | sharpen | Sharpen image | [sharpen documentation](https://help.aliyun.com/zh/oss/user-guide/sharpen) | | contrast | Adjust image contrast | [contrast documentation](https://help.aliyun.com/zh/oss/user-guide/contrast) | | flip | Flip image | [flip documentation](https://help.aliyun.com/zh/oss/user-guide/flip-image) | | segment | Perform image segmentation | See below | | remove | Perform image removal | See below | ### Basic Image Processing Basic image processing capabilities are provided by OSS. For detailed parameters and usage of each feature, please refer to the reference links in the overview table. For image watermarks in watermark processing, the watermark image's pds_schema format is required, i.e., `pds://domains/{domain_id}/drives/{drive_id}/files/{file_id}/revisions/{revision_id}`, which needs to be URL-safe base64 encoded before use. ### Watermark Processing #### Image Watermark | Feature | Parameter Format | Description | |------|---------|------| | **Image Watermark** | `image/watermark,image_{base64(pds_schema)}` | Add image watermark, watermark image must exist in PDS | | **Watermark Position** | `image/watermark,image_{...},g_{position}` | Specify watermark position: nw(top-left), north(top-center), ne(top-right), west(left-center), center(center), east(right-center), sw(bottom-left), south(bottom-center), se(bottom-right) | | **Watermark Transparency** | `image/watermark,image_{...},t_{transparency}` | Set watermark transparency, 0-100, 100 means completely opaque | | **Watermark Ratio** | `image/watermark,image_{...},p_{percent}` | Watermark percentage of original image, 1-100 | | **Watermark Horizontal Offset** | `image/watermark,image_{...},x_{offset}` | Watermark horizontal offset distance, unit: pixels | | **Watermark Vertical Offset** | `image/watermark,image_{...},y_{offset}` | Watermark vertical offset distance, unit: pixels | | **Watermark Tiling** | `image/watermark,image_{...},repeat_1` | Tile watermark across entire image | > **Watermark image pds_schema format**: `pds://domains/{domain_id}/drives/{drive_id}/files/{file_id}/revisions/{revision_id}` > > The pds_schema needs to be URL-safe base64 encoded before use. ### AI Image Processing | Feature | Parameter Format | Description | |------|---------|------| | **Auto Segmentation** | `image/segment` | Automatically identify and extract the main subject from the image | | **Point-based Segmentation** | `image/segment,points_(x_{x},y_{y})` | Extract subject at specified coordinate point, x is distance from left edge (px), y is distance from top edge (px) | | **Rectangle Segmentation** | `image/segment,boxes_(x_{x},y_{y},w_{w},h_{h})` | Extract rectangular area, x,y are starting coordinates, w is width, h is height | | **Text-based Segmentation** | `image/segment,prompt_{base64(prompt)}` | Segment based on text description, prompt is text description (e.g., "kitten"), needs base64 encoding | | **Point-based Removal** | `image/remove,points_(x_{x},y_{y})` | Remove content at specified coordinate point | | **Rectangle Removal** | `image/remove,boxes_(x_{x},y_{y},w_{w},h_{h})` | Remove rectangular area | ### Feature Combination Multiple image editing capabilities can be combined, separated by `/`, executed from left to right in order: **Note**: Only the first operation needs the `image/` prefix, subsequent operations can be written directly without the prefix. ``` image/crop,x_50,y_50,w_200,h_200/resize,w_100/sharpen,90 image/rotate,90/resize,p_150 ``` --- ## Core Workflow ### Image Editing and Save-as Save edited image to specified PDS location. #### Step 1: Construct x-pds-process parameter and save to variable **Important**: Since x-pds-process parameter contains base64 encoding (which may include special characters like `=`), parameters must be passed using variables to avoid shell parsing errors from direct hardcoding. ```bash # Generate parameter and save to variable X_PDS_PROCESS=$(python scripts/render_image_editing_process.py \ --operations "image/resize,w_200" \ --saveas \ --target-domain-id ${TARGET_DOMAIN_ID} \ --target-drive-id ${TARGET_DRIVE_ID} \ --target-file-id ${TARGET_FILE_ID} \ --target-revision-id ${TARGET_REVISION_ID} \ --file-name "edited_image.jpg") ``` **Save-as Parameter Description**: - `--saveas`: Enable save-as functionality - `--target-domain-id`: domain_id of the save-as target (required) - `--target-drive-id`: drive_id of the save-as target (required) - `--target-file-id`: target file ID or parent folder ID (required) - `--target-revision-id`: target version ID (required when overwriting existing file, leave empty when creating new file) - `--file-name`: saved file name (required) #### Step 2: Execute save-as request ```bash aliyun pds process \ --resource-type file \ --drive-id ${SOURCE_DRIVE_ID} \ --file-id ${SOURCE_FILE_ID} \ --x-pds-process "${X_PDS_PROCESS}" \ --user-agent AlibabaCloud-Agent-Skills ``` **Success Response** (HTTP 200): ```json { "drive_id": "drive_id of saved file", "file_id": "file_id of saved file", "revision_id": "revision_id of saved file version" } ``` --- ## Common Scenario Examples ### Scenario 1: Image Scaling and Save-as ```bash # Scale to width 200px, height adjusted proportionally, and save-as python scripts/render_image_editing_process.py \ --operations "image/resize,w_200" \ --saveas \ --target-domain-id "bj31216" \ --target-drive-id "1020" \ --target-file-id "parent_folder_id" \ --file-name "resized_image.jpg" # Scale to height 200px, width adjusted proportionally, and save-as python scripts/render_image_editing_process.py \ --operations "image/resize,h_200" \ --saveas \ --target-domain-id "bj31216" \ --target-drive-id "1020" \ --target-file-id "parent_folder_id" \ --file-name "resized_image.jpg" # Limit maximum width and height, and save-as python scripts/render_image_editing_process.py \ --operations "image/resize,l_200" \ --saveas \ --target-domain-id "bj31216" \ --target-drive-id "1020" \ --target-file-id "parent_folder_id" \ --file-name "resized_image.jpg" ``` ### Scenario 2: Image Rotation and Save-as ```bash # Rotate 90 degrees and save-as python scripts/render_image_editing_process.py \ --operations "image/rotate,90" \ --saveas \ --target-domain-id "bj31216" \ --target-drive-id "1020" \ --target-file-id "parent_folder_id" \ --file-name "rotated_image.jpg" # Auto-orient (based on EXIF information) and save-as python scripts/render_image_editing_process.py \ --operations "image/auto-orient,1" \ --saveas \ --target-domain-id "bj31216" \ --target-drive-id "1020" \ --target-file-id "parent_folder_id" \ --file-name "auto_oriented_image.jpg" ``` ### Scenario 3: Auto Segmentation and Save-as ```bash # Automatically identify and extract subject, and save-as python scripts/render_image_editing_process.py \ --operations "image/segment" \ --saveas \ --target-domain-id "bj31216" \ --target-drive-id "1020" \ --target-file-id "parent_folder_id" \ --file-name "segmented_image.png" ``` ### Scenario 4: Rectangle Segmentation and Save-as ```bash # Extract top-left 100x100 area, and save-as python scripts/render_image_editing_process.py \ --operations "image/segment,boxes_(x_0,y_0,w_100,h_100)" \ --saveas \ --target-domain-id "bj31216" \ --target-drive-id "1020" \ --target-file-id "parent_folder_id" \ --file-name "cropped_image.png" ``` ### Scenario 5: Text-based Segmentation and Save-as ```bash # Segment by text description, and save-as python scripts/render_image_editing_process.py \ --operations "image/segment,prompt_5bCP5aqr" \ --saveas \ --target-domain-id "bj31216" \ --target-drive-id "1020" \ --target-file-id "parent_folder_id" \ --file-name "segmented_cat.png" ``` > Note: prompt parameter needs to be base64 encoded first, for example "小猫" (kitten) base64 encoding is "5bCP5aqr" ### Scenario 6: Rectangle Area Removal and Save-as ```bash # Remove top-left 50x50 area, and save-as python scripts/render_image_editing_process.py \ --operations "image/remove,boxes_(x_0,y_0,w_50,h_50)" \ --saveas \ --target-domain-id "bj31216" \ --target-drive-id "1020" \ --target-file-id "parent_folder_id" \ --file-name "removed_image.jpg" ``` ### Scenario 7: Combined Operations and Save-as ```bash # Scale first then rotate, and save-as python scripts/render_image_editing_process.py \ --operations "image/resize,w_200" "rotate,45" \ --saveas \ --target-domain-id "bj31216" \ --target-drive-id "1020" \ --target-file-id "parent_folder_id" \ --file-name "processed_image.jpg" # Scale + sharpen + quality adjustment, and save-as python scripts/render_image_editing_process.py \ --operations "image/resize,w_200" "sharpen,100" "quality,q_80" \ --saveas \ --target-domain-id "bj31216" \ --target-drive-id "1020" \ --target-file-id "parent_folder_id" \ --file-name "processed_image.jpg" ``` ### Scenario 8: Edit and Save-as ```bash # Scale image and save-as to specified location python scripts/render_image_editing_process.py \ --operations "image/resize,w_200" \ --saveas \ --target-domain-id "bj31216" \ --target-drive-id "1020" \ --target-file-id "parent_folder_id_123" \ --file-name "resized_image.jpg" # Overwrite existing file python scripts/render_image_editing_process.py \ --operations "image/resize,w_200" \ --saveas \ --target-domain-id "bj31216" \ --target-drive-id "1020" \ --target-file-id "existing_file_id_456" \ --target-revision-id "revision_789" \ --file-name "resized_image.jpg" ``` --- ## Error Handling | HTTP Status Code | 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 editing functionality. #### 2. Insufficient Permissions ```json { "code": "ForbiddenNoPermission.file", "message": "No Permission to access resource file" } ``` **Solution**: - Ensure current user has `DownloadFile` permission for source image - Ensure current user has `DownloadFile` permission for watermark image - Ensure current user has `CreateFile` permission for save-as target location #### 3. Invalid Parameter (InvalidParameter.XPdsProcess) ```json { "code": "InvalidParameter.XPdsProcess", "message": "The input parameter x-pds-process is not valid." } ``` **Common Causes**: - Directly hardcoding x-pds-process parameter in command line, special characters like `=` in base64 encoding are incorrectly parsed by shell - Parameter contains invisible characters (such as line breaks) **Solution**: - **Use variable to pass parameter** (recommended): ```bash X_PDS_PROCESS=$(python scripts/render_image_editing_process.py \ --operations "image/resize,h_150" \ --saveas \ --target-domain-id "bj31216" \ --target-drive-id "101" \ --target-file-id "folder_id" \ --file-name "output.png") aliyun pds process \ --resource-type file \ --drive-id "101" \ --file-id "source_file_id" \ --x-pds-process "${X_PDS_PROCESS}" \ --user-agent AlibabaCloud-Agent-Skills ``` - Ensure there are no extra spaces or line breaks in the parameter - Check if base64 encoding is correct --- ## Best Practices ### 1. Operation Order Optimization Image editing operations are executed from left to right.合理安排顺序可以提高处理效率: - Crop first then scale: reduces data volume for subsequent processing - Rotate first then crop: avoids coordinate changes after rotation ### 2. Coordinate Determination Before using point-based or rectangle operations, it is recommended to obtain image dimensions first to ensure coordinate values are within valid range. --- ## FAQ **Q: How to get image dimension information?** A: You can use `aliyun pds get-file` command to get file information, the returned data includes image width and height information. ```bash aliyun pds get-file \ --drive-id \ --file-id \ --user-agent AlibabaCloud-Agent-Skills ``` response example: ```json { "file_id": "5d79206586bb5dd69fb34c349282718146c55da7", "name": "example.jpg", "image_media_metadata": { "width": 1920, "height": 1080 } } ``` **Q: What is the difference between segmentation and removal operations?** A: - **Segmentation (segment)**: Extract the main subject from the image, background becomes transparent - **Removal (remove)**: Remove content from specified area in the image, AI automatically fills the background **Q: What is the execution order when multiple operations are combined?** A: Operations are executed from left to right in the order they appear in x-pds-process. **Q: Will save-as operation modify the source file?** A: No, save-as operation creates a new file, the source file remains unchanged. --- ## Image Limitations - **Size Limit**: Only supports images within 20MB - **Format Limit**: Only supports the following formats - jpg, jpeg, bmp, png, heic, webp, tiff, avif ## Permission Requirements - Need `DownloadFile` permission for the image being edited - Need `DownloadFile` permission for watermark images - Need `CreateFile` permission for save-as target location --- FILE:references/mountapp.md # Mount App Installation Guide ## Overview mountapp is a PDS cloud drive mount plugin that supports mounting PDS cloud drive storage to local computer, allowing access to files in PDS cloud drive like local files. It supports Windows, macOS, and Linux systems. The following sections describe the download, installation, startup, and mounting process of the mount app, including: - Get plugin latest version and download URL - Download software installation package - Execute installation process - Start mount app - Complete mounting - Verify installation results - Query mount app status - Query mount configuration - Modify mount app configuration ## Prerequisites Before installing mount app, verify that aliyun-cli and PDS plugin are installed and configured correctly --- ## Workflow ### Step 1: Check if Mount App is Already Installed and Running Check if installed: - **Windows/macOS systems**: - Check if there are mountapp related files in `~/.edm/plugins/mountapp` directory - View version number via `~/.edm/plugins/mountapp/plugin.json` If already installed, view the plugin configuration in mountapp directory to get the installed version number: ```json { "id": "mountapp", "name": "mountapp", "version": "0.8.2", "manifest_version": "v2", "client_id": "GzpsX2VzKLNKNsxProd", "redirect_uri": "https://web-sv.aliyunpds.com/plugin_callback", "scripts": { "start": "chmod +x bin/start.sh;bin/start.sh ${port} ${user_id}", "install": "chmod +x bin/install.sh;bin/install.sh", "upgrade": "chmod +x bin/upgrade.sh;bin/upgrade.sh", "stop": "chmod +x bin/stop.sh;bin/stop.sh", "uninstall": "chmod +x bin/uninstall.sh;bin/uninstall.sh" } } ``` For example, the `version` field in the above configuration is the installed version number. - **Linux systems**: - Check if there are mountapp related files in `/opt/mountapp` directory - View service list via `systemctl list-units --type=service | grep mountapp` - View service status via `systemctl status mountapp` - View installed rpm packages via `rpm -qa | grep mountapp` - View installed dpkg packages via `dpkg -s mountapp` --- ### Step 2: Get Mount App Plugin Download URL Use command line tool to get the latest mount app version: ```bash aliyun pds mountapp --action get-latest-version --user-agent AlibabaCloud-Agent-Skills ``` Response format: ```json { "version": "0.8.2", "url": "https://example.com/mountapp-0.8.2.zip" } ``` **Version comparison logic**: - If installed version matches latest version, skip Step 3 and go directly to Step 6 - If installed version differs from latest version, continue with Step 3 to download latest installation package - If not installed locally, proceed to Step 3 to download latest installation package --- ### Step 3: Download Installation Package Based on current operating system and download URL from Step 2, download mount app plugin installation package to temporary directory. Different operating systems require different installation package types: - **Windows**: zip - **macOS**: zip - **Linux**: rpm **Download command example**: ```bash # Get download URL latest_info=$(aliyun pds mountapp --action get-latest-version --user-agent AlibabaCloud-Agent-Skills) download_url=$(echo "$latest_info" | jq -r '.url') version=$(echo "$latest_info" | jq -r '.version') # Download to temporary directory curl --max-time 600 -fL -o "/tmp/mountapp-${version}.zip" "$download_url" # Windows/macOS curl --max-time 600 -fL -o "/tmp/mountapp-${version}.rpm" "$download_url" # Linux ``` --- ### Step 4: Execute Installation Execute installation based on operating system and installation package type: #### Windows Installation 1. **Extract ZIP package**: ```powershell # Extract to ~/.edm/plugins/ Expand-Archive -Path "$env:TEMP\mountapp-${version}.zip" -DestinationPath "$env:USERPROFILE\.edm\plugins\" -Force ``` 2. **Install Dokan Driver**: Before installing Dokan driver, check if already installed using cmd query command: ``` sc query dokan1 ``` **Expected output**: If service status is displayed (RUNNING or STOPPED), it is installed; if service does not exist, Dokan driver needs to be installed. ```powershell # Use extracted Dokan MSI installation file $dokanInstaller = "$env:USERPROFILE\.edm\plugins\mountapp\pkg\Dokan_x64-noVC.msi" Start-Process msiexec.exe -ArgumentList "/i `"$dokanInstaller`" /qn /norestart" -Wait -Verb RunAs ``` #### macOS Installation 1. **Extract ZIP package**: ```bash # Extract to ~/.edm/plugins/ unzip -o "/tmp/mountapp-${version}.zip" -d ~/.edm/plugins/ ``` Grant execute permissions to extracted folder ```bash chmod +x ~/.edm/plugins/mountapp/bin/DasfsWorker chmod +x ~/.edm/plugins/mountapp/bin/dasd chmod +x ~/.edm/plugins/mountapp/bin/*.sh ``` 2 **Apple Silicon Special Settings**: If current machine has Apple silicon processor (M1/M2/M3, etc.), additional Apple settings need to be modified to allow system extension loading. Please refer to [Apple Official Documentation](https://support.apple.com/zh-cn/guide/mac-help/mchl768f7291/26/mac/26) to complete configuration. 3 **macFUSE Dependency Notes**: ⚠️ **Important**: macOS depends on macFUSE driver, which needs to be installed manually. macFUSE is a FUSE (Filesystem in Userspace) implementation for macOS that allows users to run their own file systems without kernel support. When installing macFUSE, it needs to match the current operating system version, otherwise compatibility issues may occur. The correspondence is as follows: Here is the macOS system version and recommended macFUSE driver version correspondence table: | macOS Version | macFUSE Version | Notes | |------------|--------------|------| | Tahoe 26.x | 5.1.2 | If macOS system is the latest version, it is recommended to download and install the latest version of macFUSE. | | Sequoia 15.x | 4.10.2 | | | Sonoma 14.x | 4.6.1 | | | Ventura 13.x | 4.6.1 | | | Monterey 12.x | 4.6.1 | | | Big Sur 11.x | 4.6.1 | | | Other older versions | 3.11.2 | | Based on current macOS version, guide users to complete installation. - If not currently installed, guide users to download and install the corresponding version of macFUSE. - If currently installed macFUSE version does not match system version, guide users to uninstall and install the corresponding version of macFUSE. After macFUSE installation, if prompted **System Extension Blocked**, follow these steps: - Click to open **Security & Privacy Preferences**, navigate to **System Settings > Privacy & Security** - In **Security** area, select Allow apps downloaded from **App Store and identified developers** - Authorize macFUSE (**Developer: Benjamin Fleischer**) to load. After modifying above settings, you may need to restart the computer. #### Linux Installation 1. **CentOS/RedHat (RPM systems)**: ```bash # Directly install RPM package sudo rpm -ivh /tmp/mountapp-${version}.rpm ``` 2. **Ubuntu/Debian (DEB systems)**: ```bash # Install conversion tool sudo apt-get install -y alien # Convert RPM to DEB cd /tmp sudo alien mountapp-${version}.rpm # Install converted DEB package sudo dpkg -i mountapp_${version}_*.deb ``` 3. **Check FUSE2 Dependency**: ⚠️ **Important**: Linux systems require fuse2 version. ```bash # Check if fuse2 is installed dpkg -l | grep fuse # Debian/Ubuntu rpm -qa | grep fuse # CentOS/RedHat # If fuse2 is not installed, install it first sudo apt-get install -y fuse # Ubuntu/Debian (fuse2.9.9) sudo yum install -y fuse # CentOS/RedHat ``` **Notes**: - Must install fuse2 (e.g., fuse2.9.9) - If system has fuse3, no need to uninstall fuse3, you can directly install fuse2, both can coexist --- ### Step 5: Verify Installation Verify installation results based on operating system: #### Windows Verification 1. **Check if files exist**: ```powershell # Check mountapp directory Test-Path "$env:USERPROFILE\.edm\plugins\mountapp" ``` **Expected output**: `True` 2. **Check Dokan service status**: ```powershell # Query Dokan service sc query dokan1 ``` **Expected output**: Should display service status (RUNNING or STOPPED) #### macOS Verification ```bash # Check mountapp directory ls -la ~/.edm/plugins/mountapp ``` **Expected output**: Should display mountapp related files and directories #### Linux Verification ```bash # Check mountapp directory ls -la /opt/mountapp ``` **Expected output**: Should display mountapp related files and directories --- ### Step 6: Start Software **Pre-start check**: First check if mount app is already running. If running, skip this step and go directly to Step 7. #### Check Method - **Windows**: View Task Manager, check if `DasfsWorker` process is running - **macOS**: Check if `DasfsWorker` process is running - **Linux**: Use command `systemctl status mountapp` to check if mountapp service is running #### Windows Start Mount App If not running, use following steps to start: 1. **Get User ID**: ```powershell $userIdJson = aliyun pds mountapp --action get-user-id --user-agent AlibabaCloud-Agent-Skills $userId = ($userIdJson | ConvertFrom-Json).user_id ``` 2. **Generate Random Port and Save**: ```powershell # Randomly select a port from range 49152~65535 $port = Get-Random -Minimum 49152 -Maximum 65536 # Write to port file (no newline) [System.IO.File]::WriteAllText("$env:USERPROFILE\.dasfs-worker-port", $port) ``` 3. **Create startup script `start-task.ps1`**: ```powershell $binDir = "$env:USERPROFILE\.edm\plugins\mountapp\bin" $logDir = "$env:USERPROFILE\.pdsdrive\log" # Ensure log directory exists New-Item -ItemType Directory -Path $logDir -Force | Out-Null # Create startup script $script = @" `$logDir = `"$logDir`" cd `"$binDir`" .\start.bat $port $userId 2>&1 | Out-File -FilePath `"`$logDir\mountapp-task.log`" -Append "@ $script | Out-File -FilePath "$binDir\start-task.ps1" -Encoding UTF8 ``` 4. **Register Windows Scheduled Task**: ```powershell $taskName = "PDS MountApp Service" # Define task action $action = New-ScheduledTaskAction ` -Execute "powershell.exe" ` -Argument "-ExecutionPolicy Bypass -WindowStyle Hidden -File `"$binDir\start-task.ps1`"" ` -WorkingDirectory $binDir # Define task principal (run as current user) $principal = New-ScheduledTaskPrincipal ` -UserId "$env:USERDOMAIN\$env:USERNAME" ` -LogonType S4U ` -RunLevel Limited # Define trigger (at system startup) $trigger = New-ScheduledTaskTrigger -AtStartup # Define task settings (auto restart on failure) $settings = New-ScheduledTaskSettingsSet ` -AllowStartIfOnBatteries ` -DontStopIfGoingOnBatteries ` -RestartCount 999 ` -RestartInterval (New-TimeSpan -Minutes 1) # Register scheduled task Register-ScheduledTask ` -TaskName $taskName ` -Action $action ` -Principal $principal ` -Trigger $trigger ` -Settings $settings ` -Force # Start task Start-ScheduledTask -TaskName $taskName ``` 5. **Verify Startup**: Wait 5-10 seconds, then check process: ```powershell Get-Process | Where-Object {$_.ProcessName -like "*DasfsWorker*"} ``` **Expected output**: Should display DasfsWorker process running --- #### macOS Start Mount App Use **launchd (plist)** method to start mount app, which is the most stable and reliable startup method on macOS. 1. **Get User ID and Generate Port**: ```bash # Get user ID user_id_json=$(aliyun pds mountapp --action get-user-id --user-agent AlibabaCloud-Agent-Skills) user_id=$(echo "$user_id_json" | jq -r '.user_id') # Generate random port port=$((49152 + RANDOM % 16384)) # Write to port file (no newline) echo -n $port > ~/.dasfs-worker-port ``` 2. **Create plist file**: Create `~/Library/LaunchAgents/com.aliyun.pds.mountapp.plist`: ```xml Label com.aliyun.pds.mountapp ProgramArguments /bin/bash -c cd $HOME/.edm/plugins/mountapp/bin && ./DasfsWorker start --port=PORT_PLACEHOLDER --userId=USERID_PLACEHOLDER --dataPath=$HOME/.pdsdrive --logPath=$HOME/.pdsdrive/log RunAtLoad KeepAlive SuccessfulExit Crashed WorkingDirectory $HOME/.edm/plugins/mountapp/bin EnvironmentVariables HOME $HOME PATH /usr/local/bin:/usr/bin:/bin:/usr/sbin:/sbin StandardOutPath $HOME/.pdsdrive/log/mountapp.out.log StandardErrorPath $HOME/.pdsdrive/log/mountapp.err.log ThrottleInterval 10 ``` **Note**: Placeholders in plist file need to be replaced: - Replace `PORT_PLACEHOLDER` with actual port number - Replace `USERID_PLACEHOLDER` with actual user ID - Replace `$HOME` with current user's home directory path 3. **Load and Start launchd Service**: ```bash # Ensure log directory exists mkdir -p ~/.pdsdrive/log # First time load and start launchctl load ~/Library/LaunchAgents/com.aliyun.pds.mountapp.plist sleep 2 launchctl start com.aliyun.pds.mountapp sleep 3 # If already loaded, unload first then reload # launchctl unload ~/Library/LaunchAgents/com.aliyun.pds.mountapp.plist 2>/dev/null # sleep 1 # launchctl load ~/Library/LaunchAgents/com.aliyun.pds.mountapp.plist # sleep 2 # launchctl start com.aliyun.pds.mountapp ``` 4. **Verify Startup**: Wait 3-5 seconds, then check if process started successfully: ```bash ps aux | grep DasfsWorker | grep -v grep ``` **Expected output**: Should display DasfsWorker process running **Note**: After using `launchctl unload` to stop service, it is recommended to wait 1-2 seconds before reloading to ensure process completely stops. --- #### Linux Start Mount App Linux starts automatically after installing rpm or deb package, no additional startup needed. The port number is already written to `~/.dasfs-worker-port` file. If manual startup is needed: ```bash # Start service sudo systemctl start mountapp # Set to start on boot sudo systemctl enable mountapp # View status systemctl status mountapp ``` --- ### Step 7: Check and Enable Mount App Feature Before mounting, you need to enable mount app feature using command line: ```bash aliyun pds mountapp --action enable-mountapp --user-agent AlibabaCloud-Agent-Skills ``` **Success output**: ```json {"mount_app_enable": "success", "domain_id": "bj123"} ``` **Notes**: 1. If enabling fails, check: - Whether PDS drive DomainId, UserID, etc. are configured - Whether the configured account has permission to enable mount app feature 2. Enabling mount app only needs to be done once. If already enabled successfully, no need to repeat --- ### Step 8: Complete Mounting #### 8.1 Query Mount App Status Query mount app status before mounting. If already mounted, skip mounting step: ```bash aliyun pds mountapp --action get-status --user-agent AlibabaCloud-Agent-Skills ``` **Example output**: ```json { "UserId": "123456", "DomainId": "bj123", "Username": "root", "MountedStatus": "MountSuc", "Message": "" } ``` **Status description**: - `MountSuc`: Already mounted successfully, no need to mount again - `Starting`: Mounting in progress, continue querying status. If not completed within 2 minutes, report mounting failure, need to remount - `Init`: Not mounted, need to execute mounting operation --- #### 8.2 Execute Mounting Operation If status is `Init`, execute mounting: ```bash # Basic mount command aliyun pds mountapp --action mount --user-agent AlibabaCloud-Agent-Skills # Linux non-root users need to specify mount-user aliyun pds mountapp --action mount --mount-user admin --user-agent AlibabaCloud-Agent-Skills ``` **Note**: For Linux systems, check current running user. If not root user, add `--mount-user` parameter. **Success output**: ```json {"domain_id": "bj123", "user_id": "123456", "message": "mount success, please check status"} ``` --- #### 8.3 Verify Mount Status After command execution succeeds, confirm mounting completion by querying status: ```bash aliyun pds mountapp --action get-status --user-agent AlibabaCloud-Agent-Skills ``` **Expected output**: ```json { "DomainId": "bj123", "Message": "", "MountedStatus": "MountSuc", "SubDomainId": "", "UserId": "123456", "Username": "user1" } ``` If `MountedStatus` is `MountSuc`, mounting is successful! --- #### 8.4 Query Mount Configuration After successful mounting, you can query mount app configuration: ```bash aliyun pds mountapp --action get-config --user-agent AlibabaCloud-Agent-Skills ``` **Example output**: ```json { "DiskCachePath": "/Users/user1/.pdsdrive/cf8833674b2544b8aeeed2426bbdc4d9/cache", "DiskCacheSize": 5, "DomainId": "bj123", "Language": "zh", "MemoryCacheSize": 64, "MountPath": "/Users/user1/PDSDrive", "MountUser": "", "ShowIconPreview": true, "SubDomainId": "", "UserId": "cf8833674b2544b8aeeed2426bbdc4d9", "Version": "0.8.2" } ``` **Important**: The `MountPath` field is the mount point path where mounting succeeded. Users can access this path to view mounted files. --- #### 8.5 About Boot Startup and Exception Handling 1. **Not mounted after boot startup**: If after boot startup, query status shows not mounted (`Init`), need to execute mounting using command `aliyun pds mountapp --action mount --user-agent AlibabaCloud-Agent-Skills` 2. **After process abnormal restart**: If process has exception and restarts, query status shows not mounted (`Init`), need to execute mounting using command `aliyun pds mountapp --action mount --user-agent AlibabaCloud-Agent-Skills` --- ### Step 9: Modify Mount App Configuration Currently supports modifying mount app language. The command: ```bash aliyun pds mountapp --action set-config --language zh --user-agent AlibabaCloud-Agent-Skills ``` Currently supports three languages: - `zh`: Chinese - `en`: English - `es`: Spanish **Note**: Changing mount language requires remounting to take effect --- ## Success Verification ### Verify Mount Point is Accessible Access mount point based on operating system: **Windows**: ```powershell # Default mount point: P:\ dir P:\ ``` **macOS/Linux**: ```bash # Default mount point: ~/PDSDrive ls -la ~/PDSDrive ``` **Expected output**: ``` Personal Space Team Space Received Shares ``` Expect at least one of the above three directories to exist ### Mount App Directory Structure After successful mounting, top-level directory is read-only, with some system-level directories: ```bash tree -L 1 ~/PDSDrive/ ~/PDSDrive/ ├── Personal Space ├── Team Space └── Received Shares ``` **Directory description**: - **Personal Space**: This directory allows direct read/write - **Team Space**: This directory is read-only, lists team spaces with permissions. After entering team space, may be read-only or read-write depending on granted permissions - **Received Shares**: This directory is read-only, lists shares with permissions. After entering share directory, may be read-only or read-write depending on granted permissions ### Access Files **Windows**: ```powershell # Access Personal Space dir "P:\Personal Space" ``` **macOS/Linux**: ```bash # Access Personal Space ls -la ~/PDSDrive/Personal\ Space ``` --- ## Stop and Uninstall **Important**: Stopping and uninstalling mount app are **high-risk operations**. Before operation, human confirmation is required: Please confirm all files opened under mount drive letter (such as P:\ on Windows) or mount directory (~/PDSDrive on macOS and Linux) have been saved and closed to avoid data loss. Only proceed with subsequent operations after human confirmation. ### How to Stop Mount App #### Windows ```powershell cd $env:USERPROFILE\.edm\plugins\mountapp\bin .\stop.bat ``` Stop and unregister Windows scheduled task ```powershell Stop-ScheduledTask -TaskName "PDS MountApp Service" -ErrorAction SilentlyContinue Unregister-ScheduledTask -TaskName "PDS MountApp Service" -Confirm:$false ``` #### macOS ```bash # Use launchctl to stop service launchctl stop com.aliyun.pds.mountapp launchctl unload ~/Library/LaunchAgents/com.aliyun.pds.mountapp.plist # Or use stop.sh script cd ~/.edm/plugins/mountapp/bin bash stop.sh ``` #### Linux ```bash sudo systemctl stop mountapp ``` --- ### How to Uninstall Mount App Plugin ⚠️ **Important**: Must stop mount app service before uninstalling #### Windows 1. **Stop service**: ```powershell cd $env:USERPROFILE\.edm\plugins\mountapp\bin .\stop.bat ``` 2. **Delete scheduled task**: ```powershell Unregister-ScheduledTask -TaskName "PDS MountApp Service" -Confirm:$false ``` 3. **Delete plugin files**: ```powershell Remove-Item -Path "$env:USERPROFILE\.edm\plugins\mountapp" -Recurse -Force ``` #### macOS 1. **Stop service**: ```bash launchctl stop com.aliyun.pds.mountapp launchctl unload ~/Library/LaunchAgents/com.aliyun.pds.mountapp.plist rm ~/Library/LaunchAgents/com.aliyun.pds.mountapp.plist ``` 2. **Delete plugin files**: ```bash rm -rf ~/.edm/plugins/mountapp ``` #### Linux 1. **Stop service**: ```bash sudo systemctl stop mountapp sudo systemctl disable mountapp ``` 2. **Uninstall software package**: **RPM systems (CentOS/RedHat)**: ```bash sudo rpm -e mountapp ``` **DEB systems (Ubuntu/Debian)**: ```bash sudo apt-get remove mountapp ``` --- ## Default Mount Points Default mount points for different operating systems: | Operating System | Default Mount Point | |---------|-----------|| | **Windows** | `P:\` | | **macOS** | `~/PDSDrive` | | **Linux** | `~/PDSDrive` | --- ## Usage Limitations Mount app maps cloud drive storage to local file system, enabling access to PDS cloud drive files like local files. However, there are some limitations: 1. **File type limitations**: - ✅ Supports upload/download of various file types - ✅ Supports access to various documents, images, videos, etc. - ❌ Does not support read/write of certain special files, such as: database files, git code repositories, svn, encrypted files, etc. 2. **Collaboration limitations**: - ❌ Does not support simultaneous multi-user editing (collaboration) - For multi-user collaborative editing, please use PDS cloud drive's online editing feature 3. **Platform limitations**: - ❌ Does not support installation on Windows systems with ARM processors (e.g., some Microsoft Surface devices using Qualcomm Snapdragon processors) 4. **Performance limitations**: - When simultaneously transferring more than 1000 files, may require longer time - When single file size exceeds 1GB, may require longer time - Recommendation: For large numbers of files or large file transfers, use enterprise cloud drive desktop client for best performance 5. **Network requirements**: - Mount app has certain requirements for network bandwidth and stability - When network bandwidth is low or network is unstable (e.g., mobile hotspot, restricted network environment), file upload/download may fail - Recommendation: When network is poor, use sync backup or enterprise cloud drive desktop client 6. **Windows 7 compatibility**: - ⚠️ Since Microsoft terminated Windows 7 support on January 14, 2020, some Windows 7 systems may not be able to use mount app - Recommendation: Upgrade system to Windows 10 or Windows 11 --- ## Error Handling ### Common Error Scenarios | Error Type | Solution | |---------|---------|| | **Download failed** | Check network connection, retry or use mirror source | | **Verification failed** | Re-download, confirm file integrity | | **Installation failed** | Check permissions, confirm dependencies are met | | **Version conflict** | Prompt user to choose version or uninstall old version | | **Startup failed** | Check if port is occupied, check for process conflicts | | **Mount failed** | Check if service is running, check if driver is installed, view log files | ### Log File Locations **Windows**: ``` %USERPROFILE%\.pdsdrive\log\mountapp-task.log ``` **macOS**: ``` ~/.pdsdrive/log/mountapp.out.log ~/.pdsdrive/log/mountapp.err.log ``` **Linux**: ```bash journalctl -u mountapp -n 50 ``` --- ## Best Practices 1. ✅ **Always complete preparation before installation** 2. ✅ **Configure PDS-specific config (domain_id, user_id, authentication-type)** 3. ✅ **Check if already installed and version before installation** 4. ✅ **Verify driver/service is running** 5. ✅ **Windows uses scheduled tasks for boot startup** 6. ✅ **macOS uses launchd (plist) for stable startup** 7. ✅ **Linux ensure fuse2 dependency is installed** 8. ✅ **Set timeout when handling "Starting" mount status** 9. ✅ **Query actual mount path (do not assume default path)** 10. ✅ **Stop service before cleanup/uninstall** --- ## Task Progress Tracking Use the following checklist to track mount app installation progress: ``` Mount app download, install, and startup progress: - [ ] Step 1: Check if mount app is installed and current version - [ ] Step 2: Get mount app plugin latest version and download URL - [ ] Step 3: Download installation package (if update or first install needed) - [ ] Step 4: Execute installation (extract/driver install/dependency install) - [ ] Step 5: Verify installation results - [ ] Step 6: Start mount app (check process/register boot startup) - [ ] Step 7: Check and enable mount app feature (enable-mountapp) - [ ] Step 8: Complete mounting (query status and execute mount command) ``` --- ## Reference Resources ### PDS CLI Plugin Extended Commands for Mount App (mountapp) Feature: | CLI Command | Description | Usage Scenario | |-------------|----------------|-------|| | `aliyun pds mountapp --action get-latest-version` | Get mount app latest version and download URL | Used in Step 2 for checking updates | | `aliyun pds mountapp --action get-user-id` | Get current user ID | Used in Step 6 to get user ID required for startup | | `aliyun pds mountapp --action enable-mountapp` | Enable mount app feature for cloud drive | Used in Step 7 to enable mount app feature | | `aliyun pds mountapp --action mount` | Execute mount operation | Used in Step 8 to mount cloud drive | | `aliyun pds mountapp --action get-status` | Query mount app status | Used in Step 8 to check mount status | | `aliyun pds mountapp --action get-config` | Query mount app configuration | Used to view current mount settings and mount point | | `aliyun pds mountapp --action update-config` | Update mount app configuration | Used to modify mount settings | **Note**: All `aliyun pds mountapp` commands must include `--user-agent AlibabaCloud-Agent-Skills` flag. ### ### Official Documentation - **PDS Cloud Drive Mount App Plugin**: https://help.aliyun.com/zh/pds/drive-and-photo-service-ent/user-guide/mount-drives?spm=a2c4g.750001.0.i2 - **Aliyun CLI Documentation**: https://help.aliyun.com/zh/cli/ 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 have obtained the drive_id to search in and need to search for files under that drive **Purpose**: Search for corresponding files and get file attributes such as file_id ## 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 category conditions from scalar query 4. Connect all parts with correct logical operators **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, e.g., "beach photos from this year" can use both time range and semantic description 3. **Note pagination limits**: limit maximum is 100, large result sets require pagination 4. **Time format specification**: Time conditions use UTC format `YYYY-MM-DDTHH:mm:ss` 5. **Language consistency in semantic search**: Semantic query text should maintain the same language as user input, do not translate 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 -*- """ 构建 PDS SearchFile API 查询字符串将标量查询 JSON 和语义查询 JSON 拼接为 SearchFile API 的 query 字符串。支持递归解析嵌套查询条件，合并 modality 和 category 条件。用法: 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: """转义查询字符串中的特殊字符（反斜杠和双引号）""" return value.replace("\\", "\\\\").replace('"', '\\"') def _format_value(field: str, value: str) -> str: """ 根据字段类型格式化值。 Args: field: 字段名 value: 字段值 Returns: 格式化后的值字符串（string/date 类型加引号，long/boolean 不加引号） """ # 获取字段类型，未知字段默认为 string field_info = field_schema.get(field.lower(), {}) field_type = field_info.get("type", "string") # long 和 boolean 类型不加引号 if field_type in ("long", "boolean"): return str(value) # string 和 date 类型加引号，并转义 escaped = _escape_value(str(value)) return f'"{escaped}"' def _parse_query_recursive(query: Dict[str, Any]) -> Tuple[str, Set[str]]: """ 递归解析 Query 对象为查询字符串。 Args: query: Query JSON 对象 Returns: (查询字符串, 提取到的 category 值集合) """ operation = query.get("Operation", "").lower() categories_found: Set[str] = set() # 操作符映射 op_map = { "lt": "<", "lte": "<=", "eq": "=", "gt": ">", "gte": ">=", "match": "match", "prefix": "prefix" } # 逻辑操作符处理 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) # 过滤空字符串（category 被移除后会产生空字符串） if sub_str: sub_parts.append(sub_str) # 过滤后判断子查询数量 if not sub_parts: return "", categories_found if operation == "not": # not 操作对子查询取反 return f"not ({sub_parts[0]})", categories_found elif len(sub_parts) == 1: # 只有一个子查询，直接返回，不需要逻辑操作符 return sub_parts[0], categories_found else: # 多个子查询，用逻辑操作符连接 joined = f" {operation} ".join(sub_parts) return f"({joined})", categories_found # 比较/匹配操作符 if operation in op_map: field = query.get("Field", "") value = query.get("Value", "") api_op = op_map[operation] # category 条件：收集并返回空字符串（在递归中移除） 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]: """ 将 modality 映射为 category。 Args: modality: 模态类型 Returns: 对应的 category 值，如果是 all 则返回 None """ mapping = { "document": "doc", "doc": "doc", "image": "image", "video": "video", "audio": "audio" } return mapping.get(modality.lower()) def build_query( scalar_json: Optional[str], semantic_json: Optional[str] ) -> Dict[str, Any]: """ 构建最终的查询结果。 Args: scalar_json: 标量查询 JSON 字符串 semantic_json: 语义查询 JSON 字符串 Returns: 包含 has_query, query, order_by, message 的结果字典 """ scalar_data = None semantic_data = None # 解析标量查询 if scalar_json: try: scalar_data = json.loads(scalar_json) except json.JSONDecodeError as e: print(f"[WARN] 标量查询 JSON 解析失败: {e}", file=sys.stderr) # 解析语义查询 if semantic_json: try: semantic_data = json.loads(semantic_json) except json.JSONDecodeError as e: print(f"[WARN] 语义查询 JSON 解析失败: {e}", file=sys.stderr) # 检查是否有有效查询 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": "很抱歉，我暂时无法理解您的搜索意图。目前支持以下搜索方式：\n1. 按文件属性搜索：如文件名、类型、大小、创建时间等\n2. 按内容语义搜索：如描述文件内容、场景等\n\n请尝试更具体地描述您想查找的文件，例如：\n- \"查找去年的PDF文档\"\n- \"海边日落的照片\"\n- \"大于10MB的视频文件\"" } query_parts: List[str] = [] all_categories: Set[str] = set() # 处理标量查询 scalar_query_str = "" if scalar_valid: result = scalar_data.get("result", {}) query_obj = result.get("Query") if query_obj: # 递归解析，同时收集 category 条件（category 在递归中被移除） scalar_query_str, cats_from_scalar = _parse_query_recursive(query_obj) all_categories.update(cats_from_scalar) # 处理语义查询 semantic_query_str = "" if semantic_valid: result = semantic_data.get("result", {}) query_text = result.get("query", "") modalities = result.get("modality", ["all"]) if query_text: escaped_text = _escape_value(query_text) semantic_query_str = f'semantic_text = "{escaped_text}"' # 从 modality 收集 category for m in modalities: cat = _modality_to_category(m) if cat: all_categories.add(cat) # 构建合并后的 category 条件 category_str = "" if all_categories: if len(all_categories) == 1: cat = list(all_categories)[0] escaped_cat = _escape_value(cat) category_str = f'category = "{escaped_cat}"' else: # 多个 category 使用 in 操作符 cats_list = sorted(list(all_categories)) escaped_cats = [f'"{_escape_value(c)}"' for c in cats_list] category_str = f'category in [{", ".join(escaped_cats)}]' # 最终拼接 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})") # 如果只有一个部分 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) # 处理排序 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="将标量查询和语义查询 JSON 拼接为 SearchFile API 的 query 字符串", formatter_class=argparse.RawDescriptionHelpFormatter, epilog=""" 示例: # 仅标量查询 python build_query.py --scalar-json '{"valid": true, "result": {"Query": {"Operation": "gte", "Field": "size", "Value": "1000"}}}' # 仅语义查询 python build_query.py --semantic-json '{"valid": true, "result": {"query": "海边日落", "modality": ["image"]}}' # 混合查询 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": "风景照片", "modality": ["image"]}}' 输出格式: { "has_query": true, "query": "拼接后的查询字符串", "order_by": "size DESC", "message": null } """ ) parser.add_argument( "--scalar-json", default=None, help="标量查询的 JSON 字符串，包含 valid 和 result 字段" ) parser.add_argument( "--semantic-json", default=None, help="语义查询的 JSON 字符串，包含 valid 和 result 字段" ) args = parser.parse_args() # 输入校验 if not args.scalar_json and not args.semantic_json: print("[INFO] 未提供任何查询参数", file=sys.stderr) # 构建查询 result = build_query(args.scalar_json, args.semantic_json) # 输出结果到 stdout 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": "description", "Value": "项目"} result, cats = _parse_query_recursive(query) results.record( "prefix 操作符", result == '(description prefix "项目")', f"期望 '(description prefix \"项目\")', 得到 '{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": ["all"]} }) 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 合并为 in", "category in" in result["query"] and "doc" in result["query"] and "video" in result["query"], f"得到 query: {result.get('query')}" ) # 语义 modality=["all"] 不添加 category semantic_json = json.dumps({ "valid": True, "result": {"query": "风景", "modality": ["all"]} }) result = build_query(None, semantic_json) results.record( "modality=all 不添加category", "category" 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": ["all"]} }) 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')}" ) # 未知字段默认加引号 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( "all -> 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 # 定义了文件元数据搜索的查询语法json结构 param_schema = { "type": "object", "properties": { "Query": { "type": "object", "$id": "#query", "description": "定义了用于文件元数据搜索的查询条件。支持嵌套的查询结构。", "properties": { "Operation": { "type": "string", "enum": ["not", "or", "prefix", "and", "lt", "match", "gte", "eq", "lte", "gt"], "description": "必需字段。指定操作的类型。可用的操作包括：\n- 逻辑操作符：and, or, not (需要 SubQueries 参数)\n- 比较操作符：lt, lte, gt, gte, eq\n- 字符串操作：prefix, match-phrase", "examples": ["and"] }, "Field": { "type": "string", "description": "指定要查询的元数据字段名。除了逻辑操作符（and, or, not）之外，所有操作都必须提供此字段。", "examples": ["Size"] }, "Value": { "type": "string", "description": "要查询的目标值。所有类型的值（包括数字和时间戳）都必须以字符串形式提供。不适用于逻辑操作符（and, or, not）。", "examples": ["10"] }, "SubQueries": { "type": "array", "description": "当操作为逻辑操作符（and, or, not）时，此字段为必需。它包含一组嵌套的查询条件，这些条件遵循父操作符的逻辑。例如，当父操作的 Operation 为 'and' 时，SubQueries 中的所有条件必须同时满足（AND 逻辑）。", "items": { "$ref": "#query" } } } }, "Sort": { "type": "string", "description": "定义排序所依据的字段，多个字段之间用逗号分隔。最多可以指定5个字段。字段的顺序决定了排序的优先级。例如：'Size,Filename'", "examples": ["Size,Filename"] }, "Order": { "type": "string", "enum": ["asc", "desc"], "description": "指定 Sort 字段中每个字段的排序顺序。可选值：\n- asc: 升序（默认值）\n- desc: 降序\n可以使用逗号分隔来为多个字段指定不同的排序顺序（例如：'asc,desc'）。顺序的数量不能超过 Sort 字段的数量。如果某个字段的顺序未指定，则默认为 'asc'。", "examples": ["asc,desc"] } }, "required": [] } # 定义了文件元数据的字段和对应的值类型 field_schema = { "parent_file_id": { "type": "string", "description": "父文件夹 ID。", "examples": ["root"] }, "name": { "type": "string", "description": "文件名（支持 match 模糊匹配）。", "examples": ["sampleobject.jpg"] }, "type": { "type": "string", "enum": ["file", "folder"], "description": "文件类型：file（文件）、folder（文件夹）。", "examples": ["file"] }, "file_extension": { "type": "string", "description": "文件后缀（不含点），如 pdf、jpg。", "examples": ["pdf"] }, "description": { "type": "string", "description": "描述（single_word分词），可短语匹配。", "examples": ["项目文档"] }, "mime_type": { "type": "string", "description": "表示文件格式的MIME类型。", "examples": ["image/jpeg"] }, "starred": { "type": "boolean", "description": "是否收藏。", "examples": ["true"] }, "created_at": { "type": "date", "description": "创建时间（UTC），格式为 2006-01-02T00:00:00。", "examples": ["2025-01-01T00:00:00"] }, "updated_at": { "type": "date", "description": "最后修改时间（UTC），格式为 2006-01-02T00:00:00。", "examples": ["2025-01-01T00:00:00"] }, "status": { "type": "string", "description": "文件状态：available（可用）。", "examples": ["available"] }, "hidden": { "type": "boolean", "description": "是否隐藏文件。", "examples": ["false"] }, "size": { "type": "long", "description": "文件大小，单位为字节（bytes）。", "examples": ["1000"] }, "image_time": { "type": "date", "description": "图片或视频的拍摄时间（来自 EXIF），格式为 2006-01-02T00:00:00。", "examples": ["2025-01-01T00:00:00"] }, "last_access_at": { "type": "date", "description": "最近一次访问的时间，格式为 2006-01-02T00:00:00。", "examples": ["2025-01-01T00:00:00"] }, "category": { "type": "string", "enum": ["image", "video", "audio", "doc", "app", "others"], "description": "文件分类：image（图片）、video（视频）、audio（音频）、doc（文档）、app（应用）、others（其他）。", "examples": ["image"] }, "label": { "type": "string", "description": "系统标签名称。", "examples": ["风景"] }, "face_group_id": { "type": "string", "description": "人脸分组ID，由分组列表接口获取，通过该字段进行查询分组下的照片。", "examples": ["group-id-xxx"] }, "address": { "type": "string", "description": "地址，只能选择一个行政等级查询，如想查国家则填'中国'，查省份则填'浙江省'，查城市则填'杭州市'，区县则填'西湖区'或'桐庐县'、街道或镇则填'西溪街道'或'三墩镇'。", "examples": ["杭州市"] } } # 标准标量查询的 json schema def get_json_schema(param_schema: dict) -> dict: json_schema = { "type": "object", "properties": { "valid": { "type": "boolean", "description": "一个布尔标志，用于指示用户的输入是否能够映射到所定义的查询模式。在以下两种情况下，该值将为 `false`：1) 输入中不包含任何针对已定义字段的可识别关键词。2) 输入中仅包含用于未在模式（schema）中定义的字段的术语（例如，按 'color' 或 'importance' 进行查询）。" }, "result": param_schema }, "required": ["valid"] } return json_schema # 描述了是否需要用标准标量查询来搜索文件，以及如何提取标量查询的参数 def schalar_search_prompt() -> str: output = f""" # 任务我需要你将自然语言转换为数据库查询参数： {json.dumps(param_schema, ensure_ascii = False, indent = None)} ## 查询字段定义 {json.dumps(field_schema, ensure_ascii = False, indent = None)} ## 查询字段校验在处理任何查询之前，你必须： 1. 仅处理明确引用到已定义字段的查询。 2. 如果输入查询违反此规则（例如不包含对任何已定义字段的引用，或只包含无法映射的术语），你必须返回一个输出，其中 "valid" 设为 false。应返回 {json.dumps({"valid": False}, ensure_ascii = False)} 的查询示例： - "乐山大佛"（未指定字段，不要假设是文件名） - "红色的"（color 不是已定义字段） - "重要文件"（importance 不是已定义字段）不应判定为无效的查询示例： - "图片文件"（category eq image） ## 查询操作使用指南每个操作都有特定的使用场景与语义： ### 数值比较操作 - 'eq'：精确相等比较（"等于"、"是"、"为"） - 'gt'：大于（"大于"、"超过"） - 'gte'：大于等于（"大于等于"、"不少于"） - 'lt'：小于（"小于"、"低于"） - 'lte'：小于等于（"小于等于"、"不超过"） ### 文本检索操作 - 'match'：用于在字段内检索特定文本内容（"包含文字"、"内容有"） - 'prefix'：用于路径/URI 前缀匹配（"目录下"、"文件夹中"）或字符串前缀匹配 ### 逻辑操作 - 'and'：所有条件均需为真（"且"、"并且"、"同时"） - 'or'：任一条件为真即可（"或者"、"或"） - 'not'：对条件取反（"不"、"非"） ## 排序与顺序使用指南 ### 排序（Sort）支持最多 5 个用逗号分隔的排序字段，字段顺序决定排序优先级。常见用法： - 单字段："size" - 多字段："size,name"（先按 size 排序，若 size 相同再按 name 排序）常见表达： - "按大小排序" -> "size" - "按拍摄时间排" -> "image_time" - "按名称排列" -> "name" - "先按大小再按时间排序" -> "size,image_time" 推荐字段组合： - "size,name"：适用于按文件大小排序，并保证同大小文件的稳定顺序 - "image_time,name"：适用于按时间顺序列出，并保证确定性的次序 - "name"：按字母顺序列出 ### 顺序（Order）用逗号分隔指定排序方向： - "asc"：升序（"升序"、"从小到大"） - "desc"：降序（"降序"、"从大到小"）如果指定的 Order 值少于 Sort 字段数，剩余字段默认使用 "asc"。例如： - Sort: "Size,FileModifiedTime,Filename", Order: "desc" -> 等价于 Order: "desc,asc,asc" ## 重要使用说明 1. 'match' 只能用于文件名（name）和描述（description）的检索 2. 'prefix' 不可用于文件名（name）的前缀匹配 3. **遵循简约原则**：构造查询时，应严格避免过度推断。只添加用户输入中明确要求的过滤条件。例如，若用户没有提及文件名，则查询中不得包含 `name` 相关的条件。此原则适用于所有字段。 """.strip() output += "\n\n" output += """ ## 示例 ### 示例 1 一些自然语言输入可能较为口语化并使用缩写。示例： "文件的mime类型为docx" 你应优先将口语化缩写转换为其完整形式： {'Query': {'Operation': 'eq', 'Field': 'mime_type', 'Value': 'application/vnd.openxmlformats-officedocument.wordprocessingml.document'}} "搜索pdf文件" 你应优先将口语化缩写转换为其完整形式： {'Query': {'Operation': 'eq', 'Field': 'extension', 'Value': 'pdf'}} ### 示例 2：时间范围查询时间表达通常意味着区间，而非单一时间点。示例： UserQueryDatetime: 2025-05-26T11:33:52+08:00 输入："6月6日创建的文件" 应转换为整天范围，且必须采用UTC时间，因此对于北京时间的6月6日的一整天，对应的UTC起始时间应为2025-06-05T16:00:00，结束时间应为2025-06-06T16:00:00。一天的起始时间（以及结束时间边界）必须始终严格为 16:00:00。不要在日界限使用非零的分钟/秒： {"Query": {"Operation":"and","SubQueries":[{"Operation":"gte","Field":"image_time","Value":"2025-06-05T16:00:00"},{"Operation":"lt","Field":"image_time","Value":"2025-06-06T16:00:00"}]}} UserQueryDatetime: 2025-05-26T11:33:52+08:00 输入："最近三个半小时访问过" 应转换为UTC时间跨度： {"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"}]}} 关键模式： - 日期 = 整天范围 - "最近" = 到当前时间为止的区间 - "之前/之后" = 带边界的区间 - 最后必须统一从北京时间转换为UTC时间，时间在“秒”之后就结束，不带任何毫秒、时区信息。 ### 示例 3：语言一致性始终保持输出与输入查询相同的语言。例如：输入："查找文件名为蛋糕的文件" 正确：{"Query": {"Operation": "eq", "Field": "name", "Value": "蛋糕"}} 错误：{"Query": {"Operation": "eq", "Field": "name", "Value": "cake"}} ### 示例 4：时间字段选择四个不同的时间字段： - image_time：用于图片的拍摄时间 * 示例："2023年夏天拍的照片" -> 使用 image_time - last_access_at：上次从云盘访问图片的时间 * 示例："昨天访问的文件" -> 使用 last_access_at - create_at：将文件上传到云盘或者在云盘中创建的时间 * 示例："昨天新建的文件" -> 使用 create_at * 示例："昨天上传的文件" -> 使用 create_at - update_at：文件最后更新的时间 * 示例："昨天更新的文件" -> 使用 update_at 关于照片拍摄时间的查询，始终优先使用 image_time。 ### 示例 5：基础排序输入："查找大于1GB的文件并按大小降序排列" {"Query": {"Operation": "gt", "Field": "size", "Value": "1073741824"}, "Sort": "size", "Order": "desc"} ### 示例 6：多字段排序输入："查找docx文件,按修改时间降序,同时按名称升序排列" {"Query": {"Operation": "eq", "Field": "mime_type", "Value": "application/vnd.openxmlformats-officedocument.wordprocessingml.document"}, "Sort": "image_time,name", "Order": "desc,asc"} ### 示例 7：纯排序（无查询条件）输入："所有文件按名称排序" {"Sort": "name"} ### 示例 8：多字段排序且使用默认顺序输入："所有文件按大小和创建时间排序" {"Sort": "size,image_time"} ### 示例 9：查询简化输入："文档文件或文本文件" {"Query": {"Operation": "eq", "Field": "category", "Value": "doc"}} """ output += "\n\n" json_schema = get_json_schema(param_schema = param_schema) output += f""" ## JSON 输出格式你的输出必须严格遵守以下 JSON 模式： ```json {json.dumps(json_schema, indent = None, ensure_ascii = False)} ``` ### 输出要求 - 你的响应必须是一个单一的、有效的 JSON 对象。 - 至关重要：不要在 JSON 对象之外添加任何说明文字、注释或其他内容。你的整个响应必须只包含该 JSON。 """.strip() return output if __name__ == "__main__": print(schalar_search_prompt()) FILE:scripts/get_semantic_query_prompt.py # 描述了如何判断是否需要用语义搜索文件，以及如何提取关键语义词 def semantic_search_prompt() -> str: output = """ # 任务你的任务是分析用户的自然语言输入，以判断其中是否包含对“语义（基于内容）搜索”的请求。如果包含，你需要将相关信息提取为指定的 JSON 格式。如果输入仅与文件属性（元数据）有关，或者过于模糊无法进行内容搜索，你需要指出这不是一个有效的语义查询。 ## 语义查询判定指南你必须决定用户的输入是否符合语义查询的标准。 ### 何时将查询判定为语义（valid: true）如果一个查询描述了文件中的“内容、概念或场景”，则它是语义的。请关注以下模式： - 基于内容的相似性搜索："查找与此文档相似的文档。" - 跨模态检索："查找与此音频片段对应的视频。" - 概念层面的模糊匹配："关于可持续能源的文件。" - 自然语言的场景描述："海上日落的照片。", "一个人弹吉他的视频。" 示例： - "照片里有狗"（描述视觉内容） - "关于人工智能的文档"（描述主题/概念） - "城市夜景"（描述场景） ### 何时判定为非语义（`valid: false`）如果查询“仅”涉及文件的“元数据或属性”，你应忽略这些并标记为无效。包括但不限于： - 涉及特定文件属性：大小、类型、日期（创建/修改）、文件名、路径 - 要求对属性进行精确匹配或范围筛选 - 过于模糊、没有可描述内容的查询示例： - "大于1MB的图片"（纯元数据：大小） - "文件名是'report.docx'的文件"（纯元数据：文件名） - "上周创建的文档"（纯元数据：日期） - "查找文件"（过于模糊） ## 混合查询处理如果一个查询同时包含语义描述与元数据筛选（例如：“查找今年创建的关于人工智能的PDF文档。”），你的任务是“只提取语义部分”，忽略元数据条件。只要存在语义部分，就判定 `valid: true`。 - 输入："查找今年创建的关于人工智能的PDF文档" - 你的关注点："关于人工智能" -> 这是语义部分。 - 你忽略："今年创建的"、"PDF文档" -> 这些是元数据。 ## 输出 JSON 格式你的响应必须是一个单一的 JSON 对象，结构如下： ```json { "type": "object", "properties": { "valid": { "type": "boolean", "description": "如果输入是有效的语义查询，则为 true，否则为 false。" }, "result": { "type": "object", "properties": { "query": { "type": "string", "description": "用于语义搜索的查询文本。" }, "modality": { "type": "array", "items": { "type": "string", "enum": ["document", "image", "video", "audio", "all"] }, "description": "指定要搜索的模态。如果未明确提及，默认为 ['all']。" } }, "required": ["query"], "description": "仅当 'valid' 为 true 时才应存在此字段。" } }, "required": ["valid"] } ``` - 如果 valid 为 false，应省略 result 字段。 - 如果未明确提及模态，默认使用 ["all"]。 ## 语义查询构造规则构造 result 对象中的 query 字符串时，严格遵循以下规则： 1. 保留完整语境：包含所有相关的上下文信息。不要缩短或简化专有名词或具体描述。 - 示例："杭州西湖的樱花" -> query: "杭州西湖的樱花"（不要仅写“樱花”） 2. 保持地理位置信息的完整性：地名需保留完整。 - 示例："北京故宫的建筑" -> query: "北京故宫的建筑" 3. 维护主体关系：当时间、地点与主体构成一个连贯概念时，要保留这些关系。 - 示例："春天的杭州西湖景色" -> query: "春天的杭州西湖景色" 4. 语言一致性（重要）：输出必须与用户输入使用相同语言。用户用中文查询，query 字段必须是中文；用户用英文查询，query 字段必须是英文。不要翻译 query 字段的内容。 ## 示例 ### 示例 1：纯语义查询用户输入：`找一些海滩日落的照片` （自我校正：用户在描述一个视觉场景“海滩日落”，并明确了模态“照片”（image）。这是一个明确的语义查询。）输出： ```json { "valid": true, "result": { "query": "海滩日落", "modality": ["image"] } } ``` ### 示例 2：纯元数据查询用户输入：`查找大于10MB的视频文件` （自我校正：该查询仅涉及文件属性（大小），未描述内容。因此，它不是语义查询。） ```json 输出： { "valid": false } ``` ### 示例 3：混合查询（语义 + 元数据）用户输入：`去年夏天拍的关于家庭聚会的照片` （自我校正：该查询包含一个元数据筛选（“去年夏天拍的”）和一个语义概念（“家庭聚会”）。按照规则，我必须忽略元数据，只提取语义部分。）输出： ```json { "valid": true, "result": { "query": "家庭聚会", "modality": ["image"] } } ``` ### 示例 4：非内容查询用户输入：`查找文件` （自我校正：该查询过于模糊，缺乏任何可描述的内容，无法进行语义搜索。）输出： ```json { "valid": false } ``` ### 示例 5：具有复杂语境的语义查询用户输入：`北京故宫的雪景` （自我校正：该查询描述了一个场景。未明确提到“照片”或“视频”等具体模态，因此 modality 默认为 ["all"]。）输出： ```json { "valid": true, "result": { "query": "北京故宫雪景", "modality": ["all"] } } ``` ### 示例 6：语言一致性（英文输入）用户输入：`Find pictures of a cat sleeping on a sofa` （自我校正：输入是英文，因此输出的 query 也必须是英文。禁止翻译。）正确输出： ```json { "valid": true, "result": { "query": "a cat sleeping on a sofa", "modality": ["image"] } } ``` 错误示例——不要这样做： ```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_image_editing_process.py #!/usr/bin/env python3 """ Image Editing Parameter Generation Script Used to generate x-pds-process parameters for PDS image editing """ import argparse import base64 from typing import Optional, List def url_safe_base64_encode(text: str) -> str: return base64.urlsafe_b64encode(text.encode()).rstrip(b'=').decode() def build_pds_schema(domain_id: str, drive_id: str, file_id: str, revision_id: str = "") -> str: """Build pds_schema format""" return f"pds://domains/{domain_id}/drives/{drive_id}/files/{file_id}/revisions/{revision_id}" def build_oss_process(action: str, params: List[str] = None) -> str: """Build OSS image processing parameters Args: action: Operation type, such as resize, rotate, crop, etc. params: Parameter list, such as ['w_100', 'h_100'] Returns: OSS processing parameter string """ if params: return f"{action},{','.join(params)}" return action def build_segment_process(segment_type: str, params: dict = None) -> str: """Build segmentation parameters Args: segment_type: Segmentation type - 'auto': Auto recognition - 'point': Point-based segmentation, requires x, y coordinates - 'box': Rectangle segmentation, requires x, y, w, h parameters - 'text': Text-based segmentation, requires prompt parameter params: Corresponding parameters Returns: Segmentation parameter string """ if segment_type == 'auto': return "image/segment" elif segment_type == 'point': x = params.get('x') y = params.get('y') return f"image/segment,points_(x_{x},y_{y})" elif segment_type == 'box': x = params.get('x') y = params.get('y') w = params.get('w') h = params.get('h') return f"image/segment,boxes_(x_{x},y_{y},w_{w},h_{h})" elif segment_type == 'text': prompt = params.get('prompt') prompt_base64 = url_safe_base64_encode(prompt) return f"image/segment,prompt_{prompt_base64}" else: raise ValueError(f"Unsupported segmentation type: {segment_type}") def build_remove_process(remove_type: str, params: dict) -> str: """Build image removal parameters Args: remove_type: Removal type - 'point': Point-based removal, requires x, y coordinates - 'box': Rectangle removal, requires x, y, w, h parameters params: Corresponding parameters Returns: Removal parameter string """ if remove_type == 'point': x = params.get('x') y = params.get('y') return f"image/remove,points_(x_{x},y_{y})" elif remove_type == 'box': x = params.get('x') y = params.get('y') w = params.get('w') h = params.get('h') return f"image/remove,boxes_(x_{x},y_{y},w_{w},h_{h})" else: raise ValueError(f"Unsupported removal type: {remove_type}") def build_watermark_process(watermark_image_schema: str, params: dict = None) -> str: """Build watermark parameters Args: watermark_image_schema: Watermark image pds_schema (before base64 encoding) params: Watermark parameters (position, transparency, etc.) Returns: Watermark parameter string """ watermark_base64 = url_safe_base64_encode(watermark_image_schema) # Basic watermark parameters watermark_str = f"image/watermark,image_{watermark_base64}" # Add other watermark parameters if params: for key, value in params.items(): watermark_str += f",{key}_{value}" return watermark_str def build_saveas_process(target_schema: str, file_name: str) -> str: """Build save-as parameters Args: target_schema: Target location pds_schema file_name: Saved file name Returns: Save-as parameter string """ schema_base64 = url_safe_base64_encode(target_schema) name_base64 = url_safe_base64_encode(file_name) return f"sys/saveas,f_{schema_base64},name_{name_base64}" def generate_x_pds_process(operations: List[str], saveas: dict = None) -> str: """Generate complete x-pds-process parameter Args: operations: Image editing operation list, executed in order. The first operation needs 'image/' prefix, subsequent operations do not need 'image/' prefix (e.g., ['image/segment', 'resize,p_200']) saveas: Save-as parameters, including target_domain_id, target_drive_id, target_file_id, target_revision_id, file_name Returns: Complete x-pds-process parameter string """ # Process operation list: ensure only the first operation has image/ prefix processed_operations = [] for i, op in enumerate(operations): if i == 0: # First operation, ensure it has image/ prefix if not op.startswith('image/'): processed_operations.append(f'image/{op}') else: processed_operations.append(op) else: # Subsequent operations, remove image/ prefix if op.startswith('image/'): processed_operations.append(op[6:]) # Remove 'image/' prefix else: processed_operations.append(op) # Combine all operations x_pds_process = "/".join(processed_operations) # Add save-as parameters if saveas: target_schema = build_pds_schema( saveas['target_domain_id'], saveas['target_drive_id'], saveas['target_file_id'], saveas.get('target_revision_id', '') ) saveas_str = build_saveas_process(target_schema, saveas['file_name']) x_pds_process = f"{x_pds_process}|{saveas_str}" return x_pds_process def main(): parser = argparse.ArgumentParser(description='Generate image editing request parameter x-pds-process') # Basic parameters parser.add_argument('--operations', required=True, nargs='+', help='Image editing operation list, multiple operations separated by spaces') # Save-as parameters (optional) parser.add_argument('--saveas', action='store_true', help='Whether to save-as the edited image') parser.add_argument('--target-domain-id', help='Target domain_id for save-as') parser.add_argument('--target-drive-id', help='Target drive_id for save-as') parser.add_argument('--target-file-id', help='Target file_id for save-as (parent_file_id when creating new file)') parser.add_argument('--target-revision-id', help='Target revision_id for save-as (empty when creating new file)') parser.add_argument('--file-name', help='File name for save-as') args = parser.parse_args() # Build save-as parameters saveas = None if args.saveas: if not all([args.target_domain_id, args.target_drive_id, args.target_file_id, args.file_name]): raise ValueError("Save-as requires target-domain-id, target-drive-id, target-file-id and file-name") saveas = { 'target_domain_id': args.target_domain_id, 'target_drive_id': args.target_drive_id, 'target_file_id': args.target_file_id, 'target_revision_id': args.target_revision_id or '', 'file_name': args.file_name } # Generate x-pds-process x_pds_process = generate_x_pds_process(args.operations, saveas) print(x_pds_process) if __name__ == '__main__': 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 Intelligent Workspace

Comments (0)